Agentic WorkflowをLangGraphで実装する

import operator
from typing import TypedDict, List, Tuple,Annotated
from langgraph.graph import StateGraph
from langgraph.graph import END

class AgentState(TypedDict):
    input: str
    answer: str    
    past_steps: Annotated[List[Tuple], operator.add]


workflow = StateGraph(AgentState)

def create_plan(state):
    print("@create_plan")
    print(state)

def execute_action(state):
    pass


def reflection(state):
    pass


def create_answer(state):
    pass


def should_continue_execute_action(state):
    pass


def should_replan(state):
    pass


workflow.add_node("planner", create_plan)
workflow.add_node("agent_executor", execute_action)
workflow.add_node("reflector", reflection)
workflow.add_node("answer_creator", create_answer)

workflow.set_entry_point("planner")

workflow.add_edge("planner", "agent_executor")
workflow.add_edge("answer_creator", END)

workflow.add_conditional_edges(
    "agent_executor",
    should_continue_execute_action,
    {
        "continue": "agent_executor",
        "end": "reflector"
    }
)

workflow.add_conditional_edges(
    "reflector",
    should_replan,
    {
        "continue": "planner",
        "end": "answer_creator"
    }
)

app = workflow.compile()

from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains.openai_functions import create_structured_output_runnable
from langchain_core.pydantic_v1 import BaseModel, Field

model = ChatOpenAI(
    model_name="gpt-4o",
    temperature=0
)

system_prompt = """あなたはソフトウェアのカスタマーサポートに特化したアシスタントです。\
ユーザーの質問に答えるために以下の指示に従って課題解決の計画を立ててください。\

# 絶対守るべき制約事項
- 全ての質問に対するサブタスクを用意すること
- サブタスクは、**具体的にどの機能やAzureリソースに対して何を知りたいのかを検索可能なクエリ形式で表現すること**
- できるだけ3ステップで計画を立てること
"""

planner_prompt = ChatPromptTemplate.from_messages(
    [("system", system_prompt),
     ("human", "{input}")]
)

class Plan(BaseModel):
    """plan for task execution"""
    steps: List[str] = Field(description="steps for task execution")


structured_llm = model.with_structured_output(Plan)
chain = planner_prompt | structured_llm
plan = chain.invoke({"input": "GCPのCloudRuntと同じことをAzureでやるには？"})