用LangGraph打造高可用生产级AI Agent

人工智能（AI）已经不再只是个时髦词，它正在改变我们解决实际问题的方式。从聊天机器人到自动化工作流，AI 智能体是这些创新的核心。但要打造一个可靠、可扩展、随时能上线的 AI 智能体可不是件容易事。这时候，LangGraph 登场了！它是 LangChain 的一个强大框架，能帮你轻松构建复杂、状态化的 AI 智能体，处理各种棘手的任务。这篇文章咱们就来聊聊如何用 LangGraph 打造生产就绪的 AI 智能体，还会通过一个实际案例，配上代码、工作流和图表，把一切讲得明明白白。

什么是 LangGraph？

LangGraph 是 LangChain 框架的扩展，专门用来创建状态化、多角色的 AI 智能体，支持循环工作流。跟传统的线性工作流（LangChain 里的“链”）不同，LangGraph 允许你构建图结构，让任务可以循环、分支，动态适应各种条件。这让它特别适合处理现实世界中那些复杂、需要灵活性的任务。

LangGraph 的核心功能

• • 状态化工作流：每一步自动保存状态，智能体可以暂停、恢复，或者从错误中恢复。
• • 人工介入（Human-in-the-Loop）：支持人类介入审批或编辑操作，确保关键任务有人把关。
• • 循环图：支持循环和条件逻辑，实现动态决策。
• • 与 LangChain 无缝集成：可以跟 LangChain 的工具和 LangSmith 配合，方便调试和监控。
• • 可扩展性：专为生产环境设计，支持持久执行和错误处理。

你可以把 LangGraph 想象成 AI 智能体的 GPS——它不只是沿着直路走，还能绕道、回头、根据“路况”（用户需求）灵活调整。

为什么用 LangGraph 打造生产就绪的 AI 智能体？

弄个能在 demo 里跑的 AI 智能体是一回事，但让它在生产环境里稳如老狗又是另一回事。生产系统得面对这些挑战：

• • 杂乱输入：用户给的信息不一定清楚或完整。
• • 边缘情况：意想不到的场景可能让系统崩掉。
• • 扩展性：得能处理成千上万的请求还不宕机。
• • 调试：出错时得知道问题出在哪儿。

LangGraph 帮你解决这些问题，提供了：

• • 细粒度控制：把任务拆成小块（节点），通过清晰的连接（边）管理。
• • 状态管理：在交互中保持上下文，智能体能“记住”正在发生的事。
• • 错误恢复：失败后能从断点继续。
• • 可观察性：搭配 LangSmith，追踪执行过程，轻松调试。

实际案例：旅行规划助手

咱们来看个实际例子：一个旅行规划助手，帮用户规划行程，实时获取航班和酒店信息，展示在网页上，还能根据用户需求通过邮件发送行程。这是个很适合用 LangGraph 的场景，因为它涉及：

• • 多步骤操作（搜索航班、酒店，发送邮件）。
• • 条件逻辑（比如只有用户要求时才发送邮件）。
• • 外部工具集成（航班和酒店的 API）。
• • 人工介入（让用户确认行程）。

工作原理

用户输入需求（比如“下周末从纽约去巴黎的行程”），助手会：

1. 1. 处理请求，提取关键信息（目的地、日期等）。
2. 2. 调用 API 获取航班和酒店信息。
3. 3. 把结果整理成用户友好的格式。
4. 4. 在网页上展示行程，询问用户是否需要邮件发送。
5. 5. 如果用户要求，就通过邮件发送行程。

LangGraph 的核心组件

在看代码之前，先搞懂 LangGraph 的几个核心部分：

• • 节点（Nodes）：单个任务或功能（比如调用 API、处理用户输入）。
• • 边（Edges）：节点之间的连接，定义工作流的走向。边可以是直接的（总去下一个节点）或条件的（根据逻辑选择下一个节点）。
• • 状态（State）：一个共享的数据结构，保存智能体的当前状态，比如用户输入、API 结果或对话历史。
• • 图（Graph）：把节点和边连起来的整体结构。

简单打个比方：节点就像工厂里的工人，边是传送带，状态是记录工作进度的剪贴板。

一步步实现

咱们用 Python 和 LangGraph 来实现这个旅行规划助手。假设你对 Python 有一定了解，也知道 LangChain 的基本用法。

第一步：配置环境

先安装所需库：

pip install langgraph langchain langchain-openai requests

设置 OpenAI（用于 LLM）和旅行 API（比如 Google Flights、Hotels）的 API 密钥，安全存储在环境变量中。

第二步：定义状态

状态是个共享数据结构，保存智能体运行所需的所有信息，比如用户输入、API 响应和生成的行程。我们用 typing_extensions 的 TypedDict 定义一个清晰、类型安全的结构。

from typing_extensions import TypedDict

class AgentState(TypedDict):
    user_request: str
    flight_options: list
    hotel_options: list
    itinerary: str
    send_email: bool

第三步：创建节点

每个节点执行一个特定任务。以下是我们助手的节点：

• • 处理请求：从用户输入中提取关键信息。
• • 获取航班：调用航班 API。
• • 获取酒店：调用酒店 API。
• • 整理行程：格式化结果。
• • 询问邮件：询问用户是否需要邮件发送行程。
• • 发送邮件：通过邮件发送行程。

节点代码如下：

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
import requests
from typing importDict

llm = ChatOpenAI(model="gpt-4", api_key="your-openai-api-key")

defprocess_request(state: AgentState) -> Dict:
    prompt = PromptTemplate(
        input_variables=["user_request"],
        template="从以下内容提取目的地、出发地和日期：{user_request}"
    )
    response = llm.invoke(prompt.format(user_request=state["user_request"]))
    # 假设响应包含解析后的数据（为简化示例）
    return {"user_request": response.content}

deffetch_flights(state: AgentState) -> Dict:
    # 模拟 API 调用（用真实的 Google Flights API 替换）
    flight_data = [{"airline": "Air France", "price": "$500", "time": "10:00 AM"}]
    return {"flight_options": flight_data}

deffetch_hotels(state: AgentState) -> Dict:
    # 模拟 API 调用（用真实的 Google Hotels API 替换）
    hotel_data = [{"hotel": "Paris Inn", "price": "$150/night"}]
    return {"hotel_options": hotel_data}

defcompile_itinerary(state: AgentState) -> Dict:
    itinerary = f"航班信息：\n{state['flight_options']}\n酒店信息：\n{state['hotel_options']}"
    return {"itinerary": itinerary}

defask_email(state: AgentState) -> str:
    # 模拟人工介入（生产环境中用 UI 或输入框）
    return"send_email"ifinput("是否通过邮件发送行程？(y/n): ") == "y"else"end"

defsend_email(state: AgentState) -> Dict:
    # 模拟邮件发送（用 SendGrid 或类似服务替换）
    print(f"正在发送行程邮件：{state['itinerary']}")
    return {"send_email": True}

第四步：构建工作流

现在定义图，连接节点和边。我们用条件边来决定是发送邮件还是结束工作流。

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("process_request", process_request)
workflow.add_node("fetch_flights", fetch_flights)
workflow.add_node("fetch_hotels", fetch_hotels)
workflow.add_node("compile_itinerary", compile_itinerary)
workflow.add_node("ask_email", ask_email)
workflow.add_node("send_email", send_email)

# 添加边
workflow.add_edge(START, "process_request")
workflow.add_edge("process_request", "fetch_flights")
workflow.add_edge("fetch_flights", "fetch_hotels")
workflow.add_edge("fetch_hotels", "compile_itinerary")
workflow.add_edge("compile_itinerary", "ask_email")
workflow.add_conditional_edges(
    "ask_email",
    lambda state: state.get("send_email", "end"),
    {"send_email": "send_email", "end": END}
)
workflow.add_edge("send_email", END)

# 编译图
agent = workflow.compile()

第五步：运行智能体

用一个示例输入测试智能体：

initial_state = {"user_request": "计划下周末从纽约去巴黎的行程"}
result = agent.invoke(initial_state)
print("最终行程：")
print(result["itinerary"])

示例输出

假设用户在询问邮件时输入“y”：

最终行程：
航班信息：
[{'airline': 'Air France', 'price': '$500', 'time': '10:00 AM'}]
酒店信息：
[{'hotel': 'Paris Inn', 'price': '$150/night'}]
正在发送行程邮件：航班信息：[{'airline': 'Air France', 'price': '$500', 'time': '10:00 AM'}] 酒店信息：[{'hotel': 'Paris Inn', 'price': '$150/night'}]

工作流图

工作流的可视化如下：

（此处为文字描述，实际生产中会有图表展示从处理用户请求到发送邮件或结束工作流的流程，基于用户选择）

为什么这个智能体是生产就绪的？

这个旅行规划助手之所以能达到生产就绪的标准，是因为：

• • 错误处理：LangGraph 的状态管理确保工作流在失败后能恢复。
• • 支持人工介入：询问邮件的节点允许用户干预。
• • 模块化：每个节点都是小而专注的任务，方便调试或替换组件（比如换用其他 API）。
• • 可扩展：LangGraph 的基础设施支持通过 LangGraph Platform 部署，处理大量用户。

你可以进一步优化，比如：

• • 用 LangSmith 进行调试和监控。
• • 通过 LangGraph Platform 部署以提高扩展性。
• • 集成真实 API（比如 SendGrid 发邮件，Google Flights 获取实时数据）。

其他现实世界的应用场景

LangGraph 在很多场景都能大放异彩：

• • 客服智能体：解答问题，检索知识库，必要时转交人工。
• • 研究助手：搜索网络，总结信息，生成报告。
• • 财务顾问：分析市场数据，推荐投资，监控投资组合。

比如，LinkedIn 用 LangGraph 驱动内部 SQL Bot，把自然语言查询翻译成 SQL，大幅提升团队效率。

打造生产就绪智能体的建议

• • 从简单开始：先搞个基础工作流，再逐步加复杂功能。
• • 用条件逻辑：利用条件边实现动态决策。
• • 测试边缘情况：模拟杂乱输入和失败场景，确保系统稳健。
• • 用 LangSmith 监控：追踪 token 使用、错误和性能。
• • 引入人工介入：对敏感任务使用 human-in-the-loop，避免出错。

总结

LangGraph 让打造超越 demo 的可靠 AI 智能体变得简单。通过把任务拆成小步骤（节点），用清晰的路径（边）连接，跟踪信息（状态），LangGraph 让复杂工作流变得易于管理。我们的旅行规划助手展示了如何把这些理念用在实际场景中，配上代码、结果和清晰的工作流图