Skip to content
🔗 分享本题
查看我的学习进度 →

LangGraph 用节点、条件边、共享状态和检查点表达 Agent 分支循环与恢复

🧠 记忆锚点:节点做事,边定流向,状态贯穿,检查点可恢复。

💡 答案要点

LangGraph = 用图结构构建有状态的Agent应用

为什么需要LangGraph?

场景LangChain(链式)LangGraph(图式)
简单对话✅ 够用❌ 过度设计
需要循环❌ 不支持✅ 原生支持
条件分支❌ 难实现✅ 轻松实现
多Agent协作❌ 复杂✅ 简洁

核心概念:

1. 图结构

python
from langgraph.graph import StateGraph, END

# 定义状态
class AgentState(TypedDict):
    messages: list
    next_action: str

# 创建图
workflow = StateGraph(AgentState)

# 添加节点(每个节点是一个函数)
workflow.add_node("researcher", research_node)
workflow.add_node("writer", write_node)
workflow.add_node("reviewer", review_node)

# 添加边(定义流程)
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")

# 条件边(根据状态决定下一步)
workflow.add_conditional_edges(
    "reviewer",
    should_continue,  # 判断函数
    {
        "continue": "writer",  # 如果需要修改,回到writer
        "end": END  # 如果通过,结束
    }
)

2. 状态管理

python
def research_node(state: AgentState):
    """研究节点"""
    query = state["messages"][-1]
    results = search_tool(query)

    # 更新状态
    return {
        "messages": state["messages"] + [results],
        "next_action": "write"
    }

def write_node(state: AgentState):
    """写作节点"""
    research_data = state["messages"][-1]
    draft = llm.generate(f"根据以下信息写文章: {research_data}")

    return {
        "messages": state["messages"] + [draft],
        "next_action": "review"
    }

3. 循环与分支

python
def should_continue(state: AgentState):
    """决定是否继续循环"""
    last_message = state["messages"][-1]

    # 让LLM评估质量
    score = llm.evaluate(last_message)

    if score > 8:
        return "end"  # 质量好,结束
    else:
        return "continue"  # 质量差,重新写

完整示例:写作Agent

展开 Python 代码示例(55 行)
python
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="qwen3.5-plus")

# 定义状态
class WritingState(TypedDict):
    topic: str
    outline: str
    draft: str
    revision_count: int

# 各个节点
def outline_node(state):
    outline = llm.invoke(f"为'{state['topic']}'创建大纲")
    return {"outline": outline.content}

def draft_node(state):
    draft = llm.invoke(f"根据大纲写文章:\n{state['outline']}")
    return {"draft": draft.content}

def review_node(state):
    review = llm.invoke(f"评估文章质量(1-10分):\n{state['draft']}")
    score = int(review.content)
    return {"revision_count": state.get("revision_count", 0) + 1}

def should_revise(state):
    if state.get("revision_count", 0) >= 3:
        return "end"  # 最多修改3次

    # 评估质量
    score = llm.invoke(f"评分(1-10): {state['draft']}")
    if int(score.content) >= 8:
        return "end"
    return "revise"

# 构建图
workflow = StateGraph(WritingState)
workflow.add_node("outline", outline_node)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)

workflow.set_entry_point("outline")
workflow.add_edge("outline", "draft")
workflow.add_edge("draft", "review")
workflow.add_conditional_edges(
    "review",
    should_revise,
    {"revise": "draft", "end": END}
)

app = workflow.compile()

# 使用
result = app.invoke({"topic": "AI Agent的未来"})

LangGraph vs AutoGPT:

特性AutoGPTLangGraph
控制力低(完全自主)高(可精确控制)
可靠性低(易跑偏)高(明确流程)
适用场景探索性任务生产环境
成本高(多次试错)可控

面试话术:

示例表达(仅在能用本人经历或可复现实验佐证时使用): "LangGraph解决了LangChain的痛点:不支持循环和复杂分支。我们用LangGraph构建了写作Agent,支持多轮迭代优化,从大纲→初稿→评审→修改,循环直到质量达标。比AutoGPT可控,比纯Prompt灵活。"