为什么选 LangGraph 而不是 LangChain

LangChain适合直线型工作流：输入→处理→输出 LangGraph适合复杂工作流：有循环、条件分支、状态管理、人工干预

LangGraph的核心优势：

状态持久化：工作流可以暂停，随时恢复

人工干预点：在关键节点等待人类确认

可视化调试：工作流图可视化

时间旅行：可以回滚到任意历史状态

核心概念

python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
State：工作流传递的状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 消息历史累积
    user_input: str
    research_result: str
    final_answer: str

最简 LangGraph Agent

python
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
llm = ChatOpenAI(model="gpt-4o")
节点函数
def think(state: AgentState):
    """AI思考节点"""
    response = llm.invoke(state['messages'])
    return {"messages": [response]}
def should_continue(state: AgentState):
    """条件路由：是否需要继续"""
    last_message = state['messages'][-1]
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "use_tools"
    return "end"
def use_tools(state: AgentState):
    """工具调用节点"""
    # 执行工具调用
    pass
构建图
graph = StateGraph(AgentState)
graph.add_node("think", think)
graph.add_node("use_tools", use_tools)
入口节点
graph.set_entry_point("think")
条件分支
graph.add_conditional_edges(
    "think",
    should_continue,
    {"use_tools": "use_tools", "end": END}
)
graph.add_edge("use_tools", "think")
编译
app = graph.compile()
运行
result = app.invoke({"messages": [{"role": "user", "content": "搜索最新AI新闻"}]})

带人工干预的工作流

python
from langgraph.checkpoint.memory import MemorySaver
添加持久化
checkpointer = MemorySaver()
编译时指定中断点
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_action"]  # 在执行动作前暂停
)
运行到中断点
thread = {"configurable": {"thread_id": "user_123"}}
result = app.invoke({...}, config=thread)
查看当前状态（让人类审核）
print("AI准备执行：", app.get_state(thread).values)
人类确认后继续
input()  # 等待确认
app.invoke(None, config=thread)  # 从断点继续

实战：带记忆的客服Agent

关键设计：

意图识别节点：判断是普通问题/投诉/退款请求

知识库查询节点：检索FAQ答案

升级节点：复杂情况路由到人工

结单节点：记录对话结果

每个节点都有明确的输入输出，状态机确保流程可追踪和可重现。

生产部署建议

使用PostgreSQL作为checkpointer（而不是内存）

为每个用户会话设置独立的thread_id

监控节点执行时间，超时处理

定期清理过期的会话状态