LangGraph 状态机实战：使用图结构编排复杂 AI Agent 工作流与多轮对话系统

使用 Python 的 TypedDict 定义强类型状态：

from typing import TypedDict, Literal

class AgentState(TypedDict):
    """Agent 执行状态"""
    request: str           # 用户请求
    draft: str            # 生成的草稿
    critique: str         # 评估意见
    should_refine: bool   # 是否需要优化
    final_output: str     # 最终输出

生成节点负责根据用户请求创建初始草稿：

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

def generate_node(state: AgentState) -> AgentState:
    """生成初始草稿"""
    prompt = f"请针对以下请求生成详细回答：\n\n{state['request']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"draft": response.content, "should_refine": True}

评估节点扮演"审查者"角色，检查草稿质量：

def critique_node(state: AgentState) -> AgentState:
    """评估草稿质量并给出改进建议"""
    prompt = f"""请评估以下回答的质量：

请求：{state['request']}
回答：{state['draft']}

评估标准：
1. 是否准确回答了问题？
2. 是否包含具体示例或代码？
3. 逻辑是否清晰、结构是否完整？

如果回答质量优秀（满足以上所有标准），请输出"PASSED"。
否则，请指出需要改进的具体问题。"""

    response = llm.invoke([HumanMessage(content=prompt)])
    critique = response.content

    return {
        "critique": critique,
        "should_refine": "PASSED" not in critique
    }

当评估未通过时，优化节点根据反馈意见改进草稿：

def refine_node(state: AgentState) -> AgentState:
    """根据评估意见优化草稿"""
    prompt = f"""请根据以下评估意见改进回答：

原始请求：{state['request']}
原始回答：{state['draft']}
评估意见：{state['critique']}

请针对评估中指出的问题进行改进，生成更高质量的回答。"""

    response = llm.invoke([HumanMessage(content=prompt)])
    return {"draft": response.content, "final_output": response.content}

路由函数决定下一步执行哪个节点，实现条件分支：

def router(state: AgentState) -> Literal["refine", "finalize"]:
    """根据评估结果决定下一步"""
    if state["should_refine"]:
        return "refine"
    return "finalize"

将所有组件组装成完整的图结构：

from langgraph.graph import StateGraph, END

# 创建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("generate", generate_node)
graph.add_node("critique", critique_node)
graph.add_node("refine", refine_node)

# 设置入口
graph.set_entry_point("generate")

# 添加边
graph.add_edge("generate", "critique")
graph.add_conditional_edges(
    "critique",
    router,
    {
        "refine": "refine",
        "finalize": END
    }
)
graph.add_edge("refine", "critique")  # 优化后重新评估

# 编译
app = graph.compile()

运行 Agent 并观察每步状态变化：

# 执行并流式输出每步状态
for step in app.stream({
    "request": "解释 Python 中的装饰器原理",
    "draft": "",
    "critique": "",
    "should_refine": False,
    "final_output": ""
}):
    print(step)
    print("---")

# 获取最终结果
result = app.invoke({...})
print(result["final_output"])

使用检查点实现状态持久化，支持故障恢复和审计：

from langgraph.checkpoint.memory import MemorySaver

# 创建内存检查点
checkpointer = MemorySaver()

# 编译时传入
app = graph.compile(checkpointer=checkpointer)

# 使用线程 ID 隔离会话
config = {"configurable": {"thread_id": "session-001"}}

# 执行（状态自动保存）
result = app.invoke({...}, config=config)

# 恢复历史状态
history = app.get_state_history(config)
for state in history:
    print(state.values)