LangGraphのソースコードから見る今更聞けないAIエージェント実装（create_react_agent編）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from typing import Literal
from langchain_core.tools import tool

@tool
def get_cherry_blossom_forecast(city: Literal["札幌", "新潟", "仙台", "東京", "名古屋", "大阪", "福岡"]):
    """2025年のさくらの開花予想または実際の開花日を返します。"""
    if city == "札幌":
        return "札幌のさくら開花予想日は4/23日です。"
    elif city == "新潟":
        return "新潟のさくら開花予想日は4/3日です。"
    elif city == "仙台":
        return "仙台のさくら開花予想日は3/29日です。"
    elif city == "東京":
        return "東京のさくら開花日は3/24日です。"
    elif city == "名古屋":
        return "名古屋のさくら開花予想日は3/25日です。"
    elif city == "大阪":
        return "大阪のさくら開花予想日は3/25日です。"
    elif city == "福岡":
        return "福岡のさくら開花予想日は3/25日です。"
    else:
        return "情報がありません。"

@tool
def get_myaddress():
    """ユーザの居住地を返します"""
    return "東京"

1

tools = [get_cherry_blossom_forecast,get_myaddress]

1
2
3

from langgraph.prebuilt import create_react_agent
model = ChatOpenAI(model="gpt-4o", temperature=0)
graph = create_react_agent(model, tools=tools)

1
2

input = {"messages":["この辺りの桜の開花日は？"]}
graph.invoke(input)['messages'][-1].content

1

東京のさくら開花日は3/24日です。

1

graph.invoke({"messages":["大阪の桜の開花日は？"]})['messages'][-1].content

1

大阪のさくらの開花予想日は3/25日です。

1

graph.invoke({"messages":["ここはどこ？"]})['messages'][-1].content

1

東京です。

1

graph.invoke({"messages":["昨日の天気は？"]})['messages'][-1].content

1
2

過去の天気情報は提供していません。  
現在の場所の桜の開花予想を見ますか？ 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

if state_schema is not None:
    required_keys = {"messages", "remaining_steps"}
    if response_format is not None:
        required_keys.add("structured_response")

    schema_keys = set(get_type_hints(state_schema))
    if missing_keys := required_keys - set(schema_keys):
        raise ValueError(f"Missing required key(s) {missing_keys} in state_schema")

if state_schema is None:
    state_schema = (
        AgentStateWithStructuredResponse
        if response_format is not None
        else AgentState
    )

1
2
3
4
5
6
7
8

class AgentState(TypedDict):
    """The state of the agent."""

    messages: Annotated[Sequence[BaseMessage], add_messages]

    is_last_step: IsLastStep

    remaining_steps: RemainingSteps

1

config = {"recursion_limit": 10}

1

{'is_last_step': False, 'messages': [HumanMessage(content='xxx', additional_kwargs={}, response_metadata={}, id='xxx')], 'remaining_steps': 9}

1
2
3
4
5
6
7

if isinstance(tools, ToolNode):
    tool_classes = list(tools.tools_by_name.values())
    tool_node = tools
else:
    tool_node = ToolNode(tools)
    # get the tool functions wrapped in a tool class from the ToolNode
    tool_classes = list(tool_node.tools_by_name.values())

1
2
3
4
5
6
7
8
9
10
11
12

# Add a structured output node if response_format is provided
if response_format is not None:
    workflow.add_node(
        "generate_structured_response",
        RunnableCallable(
            generate_structured_response, agenerate_structured_response
        ),
    )
    workflow.add_edge("generate_structured_response", END)
    should_continue_destinations = ["tools", "generate_structured_response"]
else:
    should_continue_destinations = ["tools", END]

1
2
3
4
5
6
7
8
9

# We now add a conditional edge
workflow.add_conditional_edges(
    # First, we define the start node. We use `agent`.
    # This means these are the edges taken after the `agent` node is called.
    "agent",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
    path_map=should_continue_destinations,
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# Define the function that determines whether to continue or not
def should_continue(state: StateSchema) -> Union[str, list]:
    messages = _get_state_value(state, "messages")
    last_message = messages[-1]
    # If there is no function call, then we finish
    if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
        return END if response_format is None else "generate_structured_response"
    # Otherwise if there is, we continue
    else:
        if version == "v1":
            return "tools"
        elif version == "v2":
            tool_calls = [
                tool_node.inject_tool_args(call, state, store)  # type: ignore[arg-type]
                for call in last_message.tool_calls
            ]
            return [Send("tools", [tool_call]) for tool_call in tool_calls]

1
2
3
4

if should_return_direct:
    workflow.add_conditional_edges("tools", route_tool_responses)
else:
    workflow.add_edge("tools", "agent")

1
2
3
4
5
6
7
8
9
10
11

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
return workflow.compile(
    checkpointer=checkpointer,
    store=store,
    interrupt_before=interrupt_before,
    interrupt_after=interrupt_after,
    debug=debug,
    name=name,
)