LangGraph：构建具有状态与循环的 AI Agent

4.1 代码实现

from typing import Annotated, Literal, TypedDict

from langchain_core.messages import HumanMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

# 为智能体定义工具
@tool
def search(query: str):
    """调用以浏览网络。"""
    # 这是一个占位符，实际应用中应替换为真实的搜索 API
    if "sf" in query.lower() or "san francisco" in query.lower():
        return "现在是 60 度，有雾。"
    return "现在是 90 度，晴天。"

tools = [search]
tool_node = ToolNode(tools)

model = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0).bind_tools(tools)

# 定义决定是否继续的函数
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    # 如果 LLM 进行工具调用，那么我们将路径设置为 "tools" 节点
    if last_message.tool_calls:
        return "tools"
    # 否则，我们停止（回复用户）
    return END

# 定义调用模型的函数
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # 我们返回一个列表，因为这将被添加到现有列表中
    return {"messages": [response]}

# 定义一个新图
workflow = StateGraph(MessagesState)

# 定义我们将循环的两个节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 将入口点设置为 `agent`
# 这意味着这个节点是首先被调用的
workflow.set_entry_point("agent")

# 我们现在添加一个条件边
workflow.add_conditional_edges(
    # 首先，我们定义起始节点。我们使用 `agent`。
    # 这意味着这些边是在调用 `agent` 节点后执行的。
    "agent",
    # 接下来，我们传入决定下一个被调用节点的函数。
    should_continue,
)

# 我们现在从 `tools` 到 `agent` 添加一条普通边。
# 这意味着在 `tools` 被调用后，`agent` 节点会接着被调用。
workflow.add_edge("tools", 'agent')

# 初始化内存以在图运行之间保存状态
checkpointer = MemorySaver()

# 最后，我们编译它！
# 这将其编译为一个 LangChain 可运行体，
# 这意味着你可以像使用其他可运行体一样使用它。
# 请注意，我们在编译图时（可选地）传递了内存
app = workflow.compile(checkpointer=checkpointer)

# 使用可运行体
final_state = app.invoke(
    {"messages": [HumanMessage(content="sf 的天气如何")]},
    config={"configurable": {"thread_id": 42}}
)
print(final_state["messages"][-1].content)

4.2 代码详解

1. 工具定义：使用 @tool 装饰器定义了一个简单的 search 函数。在实际场景中，这里可以对接真实的搜索引擎 API。
2. 状态管理：使用 MessagesState 作为默认状态类型，它维护了一个消息列表。每个节点返回的消息会被追加到这个列表中。
3. 条件路由：should_continue 函数检查最后一条消息是否包含工具调用。如果有，流程进入 tools 节点；如果没有，流程结束（END）。
4. 循环结构：通过 add_edge("tools", 'agent')，工具执行完毕后会自动回到 agent 节点，让 LLM 根据工具结果生成最终回答。这就形成了一个循环。
5. 持久化：通过 MemorySaver() 和 thread_id，我们可以保留对话历史。即使程序重启，只要 thread_id 相同，上下文依然存在。

4.3 上下文保持示例

现在当我们传递相同的 thread_id 时，会通过保存的状态（即存储的消息列表）保留对话上下文：

final_state = app.invoke(
    {"messages": [HumanMessage(content="那纽约呢")]},
    config={"configurable": {"thread_id": 42}}
)
print(final_state["messages"][-1].content)

这将输出关于纽约天气的回答，并且 LLM 能够理解这是在旧金山之后的后续问题。

5. 持久化与人类在环

LangGraph 的持久性不仅限于保存消息历史。它允许你在图执行过程中的任何一点暂停。这对于'人类在环'场景非常有用。

5.1 暂停与恢复

你可以通过配置检查点器来启用暂停。当图执行到达某个检查点时，它会等待外部信号（例如用户确认）才能继续。这在需要人工审核敏感操作（如发送电子邮件、转账）的场景中至关重要。

5.2 错误恢复

由于状态被持久化，如果图执行失败，你可以从失败的节点之前的状态恢复，而不是从头开始。这大大提高了系统的鲁棒性。

6. 总结

LangGraph 为构建复杂的 LLM 应用提供了强大的图状控制能力。通过引入循环、状态管理和持久化机制，它解决了传统链式调用无法处理的复杂交互问题。无论是构建简单的问答机器人，还是复杂的多智能体协作系统，LangGraph 都是一个值得深入研究的工具。

在实际开发中，建议结合 LangChain 生态中的其他组件（如向量数据库、提示词模板等）一起使用，以发挥最大效能。同时，注意在生产环境中选择合适的数据后端来替代 MemorySaver，以确保数据的长期安全性和一致性。