LangGraph 入门与实战：基于 Agent 状态机的工具调用实践

定义 Agent 的状态

LangGraph 中最基础的类型是 StatefulGraph，这种图会在每一个 Node（节点）之间传递不同的状态信息。每个节点会根据自己定义的逻辑更新这个状态。

具体来说，我们需要继承 TypedDict 类来定义状态结构。下图展示了我们定义的四个关键变量：

• input: 输入字符串，代表用户的主要请求。
• chat_history: 之前的对话信息列表，作为上下文传入。
• agent_outcome: 来自代理的响应，可以是 AgentAction（需要调用工具）或 AgentFinish（任务完成）。如果是 AgentFinish，AgentExecutor 结束；否则调用请求的工具。
• intermediate_steps: 代理在一段时间内采取的行动和相应观察结果的列表。每次迭代都会更新。

这里特别需要注意的是 intermediate_steps 使用了 operator.add 注解。这表示对该状态的修改操作是追加到现有值上，而不是覆盖它，这对于维护对话历史至关重要。

class AgentState(TypedDict):
    # The input string
    input: str
    # The list of previous messages in the conversation
    chat_history: list[BaseMessage]
    # The outcome of a given call to the agent
    # Needs `None` as a valid type, since this is what this will start as
    agent_outcome: Union[AgentAction, AgentFinish, None]
    # List of actions and corresponding observations
    # Here we annotate this with `operator.add` to indicate that operations to
    # this state should be ADDED to the existing values (not overwrite it)
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

定义图中的节点

在 LangGraph 中，节点通常是一个函数或者 LangChain 中的 Runnable 对象。

我们这里定义两个核心节点：

1. agent 节点: 决定执行什么样的行动。它接收当前状态，调用 LLM 生成下一步动作。
2. action 节点: 当 agent 节点选择执行某个行动时，去调用相应的工具。

此外，还需要定义节点之间的连接，也就是边。包括条件判断的边（决定走向）和普通边（固定流向）。

# This a helper class we have that is useful for running tools
# It takes in an agent action and calls that tool and returns the result
tool_executor = ToolExecutor(tools)

# Define the agent node function
def run_agent(data):
    agent_outcome = agent_runnable.invoke(data)
    return {"agent_outcome": agent_outcome}

# Define the function to execute tools
def execute_tools(data):
    # Get the most recent agent_outcome - this is the key added in the `agent` above
    agent_action = data["agent_outcome"]
    print("agent action:{}".format(agent_action))
    output = tool_executor.invoke(agent_action[-1])
    return {"intermediate_steps": [(agent_action[-1], str(output))]}

# Define logic that will be used to determine which conditional edge to go down
def should_continue(data):
    # If the agent outcome is an AgentFinish, then we return `exit` string
    # This will be used when setting up the graph to define the flow
    if isinstance(data["agent_outcome"], AgentFinish):
        return "end"
    # Otherwise, an AgentAction is returned
    # Here we return `continue` string
    # This will be used when setting up the graph to define the flow
    else:
        return "continue"

定义图结构

定义完节点和逻辑后，就可以构建整个图了。值得注意的是，条件判断的边和普通边的添加方式是不一样的。

1. 创建 StateGraph 实例。
2. 添加节点。
3. 设置入口点（Entry Point）。
4. 添加条件边（Conditional Edges）：根据 should_continue 函数的返回值决定下一步。
5. 添加普通边（Normal Edges）：指定固定的流转顺序。
6. 编译图（Compile）：将图编译为 LangChain Runnable 对象以便运行。

# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/openai-tools-agent")

# Construct the OpenAI Functions agent
agent_runnable = create_openai_tools_agent(chat_zhipu, tools, prompt)

# Define a new graph
workflow = StateGraph(AgentState)

# Define the two nodes we will cycle between
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")

# We now add a conditional edge
workflow.add_conditional_edges(
    # First, we define the start node. We use `agent`.
    "agent",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
    # Finally we pass in a mapping.
    # The keys are strings, and the values are other nodes.
    # END is a special node marking that the graph should finish.
    {
        # If `tools`, then we call the tool node.
        "continue": "action",
        # Otherwise we finish.
        "end": END,
    },
)

# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()

运行示例

最后，我们需要准备输入数据并运行应用。输入应包含用户的初始请求和空的对话历史。

inputs = {"input": "what is the weather in NewYork", "chat_history": []}
result = app.invoke(inputs)
print(result["agent_outcome"].messages[0].content)

执行流程解析：

1. 程序启动，进入 agent 节点。
2. run_agent 函数被调用，LLM 分析输入 "what is the weather in NewYork"。
3. LLM 识别出需要获取天气信息，返回 AgentAction 指向 get_current_temperature 工具。
4. should_continue 检测到不是 AgentFinish，返回 "continue"。
5. 图流转至 action 节点，execute_tools 被调用，实际请求天气 API。
6. 工具返回结果，更新 intermediate_steps 状态。
7. 普通边将流程导回 agent 节点。
8. run_agent 再次被调用，LLM 结合历史步骤生成最终回答。
9. LLM 返回 AgentFinish，should_continue 返回 "end"。
10. 流程结束，输出最终结果。

最佳实践与注意事项

在实际开发 LangGraph 应用时，建议遵循以下最佳实践：

1. 状态管理：尽量保持状态字典的简洁。避免在状态中存储过大的原始数据，只存储必要的引用或摘要，以减少 Token 消耗和内存占用。
2. 错误处理：在 execute_tools 等节点中加入 try-except 块，防止因工具调用失败导致整个图崩溃。可以将错误信息记录到 intermediate_steps 中供 LLM 参考。
3. 超时控制：对于外部 API 调用（如天气查询），务必设置合理的超时时间，避免无限等待。
4. 调试技巧：LangGraph 支持可视化图结构。可以使用 app.get_graph().draw_mermaid_png() 来生成流程图，帮助理解复杂的节点流转关系。
5. 版本兼容性：LangGraph 和 LangChain 更新频繁，请留意官方文档的变更日志，特别是关于 StateGraph 和 Runnable 接口的变动。

总结

本文详细介绍了 LangGraph 的核心概念及实战应用。通过定义状态、节点和边，我们成功构建了一个具备工具调用能力的 Agent。LangGraph 的强大之处在于其灵活的状态管理和循环处理能力，这使得它能够胜任比简单链式调用更复杂的任务场景。掌握 LangGraph 将为构建企业级 AI 应用打下坚实基础。