LangGraph：构建具有状态的多智能体工作流框架

4. 实战案例：搜索工具智能体

下面是一个使用 LangGraph 构建简单智能体的完整示例。该智能体具备调用外部搜索工具的能力，并能根据搜索结果回答问题。

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):
    """调用以浏览网络。"""
    # 这是一个占位符，模拟搜索行为
    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)

4.2 运行与上下文保持

使用可运行体发起请求。注意 thread_id 的使用，它用于区分不同的对话会话。

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

输出结果将显示智能体根据模拟搜索结果生成的回答。

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

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

由于使用了 MemorySaver，智能体记得之前的对话历史，能够理解'那纽约呢'是对比旧金山天气的后续提问。

5. 高级特性详解

5.1 人类在环（Human-in-the-Loop）

在生产环境中，直接让 AI 执行操作可能存在风险。LangGraph 允许在特定节点插入人工审批步骤。通过配置 interrupt_before 或 interrupt_after，可以在执行前暂停流程，等待用户确认后再继续。这对于金融交易、代码部署等敏感场景尤为重要。

5.2 流式输出

LangGraph 支持在节点生成过程中实时流式传输数据。这不仅适用于文本生成，也适用于中间状态的更新。结合前端技术，可以实现打字机效果，显著提升交互体验。

5.3 子图与模块化

对于复杂的应用，可以将一部分逻辑封装成独立的子图（Subgraph）。主图通过调用子图来复用逻辑，这使得系统架构更加清晰，易于维护和扩展。

6. 总结

LangGraph 为大模型应用开发提供了强大的图状编排能力。通过引入状态管理、循环控制和持久化机制，它解决了传统 LLM 应用难以维护长上下文和复杂逻辑的问题。无论是构建简单的问答机器人，还是复杂的多智能体协作系统，LangGraph 都是一个值得深入研究的工具。开发者应充分利用其提供的检查点器和条件边功能，构建安全、可靠且可扩展的智能体应用。

在实际项目中，建议结合 LangSmith 进行监控和调试，以便及时发现并解决潜在问题。随着技术的演进，LangGraph 有望成为构建下一代 AI 应用的基础设施之一。