多Agent协作框架LangGraph进阶完全教程

适用读者：AI工程师、后端开发者、架构师 预计阅读时间：30分钟 最后更新：2025年 前置知识：Python基础、LangChain基础概念、LLM基本原理

目录

1. 引言：从单Agent到多Agent协作
2. LangGraph架构深度剖析
3. 状态图设计模式
4. 条件路由与分支
5. 多Agent编排策略
6. 人机协作节点
7. 持久化状态与检查点
8. 流式输出
9. 与CrewAI/AutoGen对比
10. 生产部署
11. 复杂工作流实战：智能客服系统
12. 复杂工作流实战：自动化研究助手
13. 性能优化与调试
14. 最佳实践与常见陷阱

1. 引言：从单Agent到多Agent协作

1.1 单Agent的局限性

当任务复杂度超过一定阈值时，单个Agent会面临以下问题：

• 上下文窗口限制：复杂任务需要大量上下文，单Agent难以容纳
• 能力边界：一个Prompt难以覆盖所有专业领域
• 调试困难：单体Agent的行为不透明，难以定位错误
• 扩展性差：新增能力需要重写整个系统Prompt

1.2 多Agent协作的优势

单Agent模式：
┌──────────────────────────────────┐
│  用户 → [全能Agent] → 输出       │
│         (一个巨大的Prompt)        │
└──────────────────────────────────┘

多Agent协作模式（LangGraph）：
┌──────────────────────────────────────────┐
│  用户 → [路由Agent] ──→ [研究Agent]      │
│              │                ↓           │
│              ├──→ [写作Agent] ←──         │
│              │                ↓           │
│              └──→ [审核Agent] → 输出      │
└──────────────────────────────────────────┘

1.3 LangGraph的核心理念

LangGraph是LangChain团队推出的有状态、多Agent编排框架，其核心理念是：

• 图结构：将工作流建模为有向图，节点是处理步骤，边是流转逻辑
• 状态驱动：所有节点共享一个可变状态对象，通过状态传递信息
• 可控流转：支持条件分支、循环、并行执行等复杂流转模式
• 持久化：内置检查点机制，支持暂停/恢复、时间旅行调试

2. LangGraph架构深度剖析

2.1 核心组件

2.2 基础架构图

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

# 1. 定义状态结构
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表（自动追加）
    current_step: str                         # 当前步骤
    results: dict                             # 各步骤结果
    user_intent: str                          # 用户意图

# 2. 创建状态图
graph = StateGraph(AgentState)

# 3. 定义节点函数
def analyze_intent(state: AgentState) -> dict:
    """分析用户意图"""
    messages = state["messages"]
    # 调用LLM分析意图
    intent = llm.invoke(f"分析以下请求的意图: {messages[-1].content}")
    return {"user_intent": intent, "current_step": "intent_analyzed"}

def process_request(state: AgentState) -> dict:
    """处理请求"""
    intent = state["user_intent"]
    # 根据意图处理
    result = handle_by_intent(intent)
    return {"results": {"main": result}, "current_step": "completed"}

# 4. 注册节点
graph.add_node("analyze_intent", analyze_intent)
graph.add_node("process_request", process_request)

# 5. 定义边（流转关系）
graph.add_edge(START, "analyze_intent")
graph.add_edge("analyze_intent", "process_request")
graph.add_edge("process_request", END)

# 6. 编译图
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

2.3 状态管理详解

LangGraph的状态管理是其最核心的设计：

from typing import TypedDict, Annotated, Union
from operator import add

# 状态可以使用多种reducer策略
class ComplexState(TypedDict):
    # 策略1: 列表追加（消息历史）
    messages: Annotated[list, add_messages]
    
    # 策略2: 数值累加
    total_tokens: Annotated[int, add]
    
    # 策略3: 字典合并
    agent_outputs: Annotated[dict, lambda a, b: {**a, **b}]
    
    # 策略4: 覆盖（默认行为）
    current_step: str
    
    # 策略5: 自定义reducer
    errors: Annotated[list, lambda a, b: a + b if len(a + b) < 10 else b]

3. 状态图设计模式

3.1 线性流水线模式

最简单的模式，节点按顺序依次执行：

# 线性流水线: A → B → C → D
graph = StateGraph(PipelineState)

graph.add_node("fetch_data", fetch_data)
graph.add_node("clean_data", clean_data)
graph.add_node("analyze", analyze)
graph.add_node("generate_report", generate_report)

graph.add_edge(START, "fetch_data")
graph.add_edge("fetch_data", "clean_data")
graph.add_edge("clean_data", "analyze")
graph.add_edge("analyze", "generate_report")
graph.add_edge("generate_report", END)

适用场景：ETL流程、文档处理流水线、顺序审批流程

3.2 路由分发模式

根据输入类型分发到不同的处理分支：

from typing import Literal

def route_by_type(state: PipelineState) -> Literal["text_processor", "image_processor", "code_processor"]:
    """根据输入类型路由"""
    input_type = state.get("input_type", "text")
    routing_map = {
        "text": "text_processor",
        "image": "image_processor",
        "code": "code_processor",
    }
    return routing_map.get(input_type, "text_processor")

graph = StateGraph(PipelineState)

# 注册处理节点
graph.add_node("classifier", classify_input)
graph.add_node("text_processor", process_text)
graph.add_node("image_processor", process_image)
graph.add_node("code_processor", process_code)
graph.add_node("aggregator", aggregate_results)

# 条件路由
graph.add_edge(START, "classifier")
graph.add_conditional_edges("classifier", route_by_type)

# 所有分支汇合
graph.add_edge("text_processor", "aggregator")
graph.add_edge("image_processor", "aggregator")
graph.add_edge("code_processor", "aggregator")
graph.add_edge("aggregator", END)

3.3 循环重试模式

当处理结果不满足条件时，重新执行：

def should_retry(state: AgentState) -> Literal["retry", "finish"]:
    """判断是否需要重试"""
    quality_score = state.get("quality_score", 0)
    retry_count = state.get("retry_count", 0)
    
    if quality_score >= 0.8:
        return "finish"
    elif retry_count >= 3:
        return "finish"  # 最多重试3次
    else:
        return "retry"

graph = StateGraph(AgentState)

graph.add_node("generate", generate_content)
graph.add_node("evaluate", evaluate_quality)

graph.add_edge(START, "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges(
    "evaluate",
    should_retry,
    {
        "retry": "generate",    # 质量不够 → 重新生成
        "finish": END,          # 质量达标 → 结束
    }
)

3.4 并行扇出/汇聚模式

多个节点并行执行，结果汇聚后继续：

import asyncio

async def parallel_research(state: ResearchState) -> dict:
    """并行执行多个研究任务"""
    tasks = [
        search_web(state["query"]),
        search_papers(state["query"]),
        search_code(state["query"]),
    ]
    results = await asyncio.gather(*tasks)
    return {
        "web_results": results[0],
        "paper_results": results[1],
        "code_results": results[2],
    }

# 在LangGraph中，可以使用Send API实现真正的并行
from langgraph.types import Send

def route_parallel(state: ResearchState) -> list:
    """将任务分发到多个并行节点"""
    return [
        Send("web_searcher", {"query": state["query"]}),
        Send("paper_searcher", {"query": state["query"]}),
        Send("code_searcher", {"query": state["query"]}),
    ]

4. 条件路由与分支

4.1 基于LLM的智能路由

from langchain_core.messages import HumanMessage, SystemMessage

class RouterState(TypedDict):
    messages: Annotated[list, add_messages]
    route: str
    complexity: str

async def intelligent_router(state: RouterState) -> dict:
    """使用LLM判断请求复杂度和路由"""
    system_prompt = """你是一个请求路由器。分析用户请求，返回JSON格式：
    {
        "complexity": "simple" | "medium" | "complex",
        "domain": "general" | "technical" | "creative" | "analysis",
        "requires_tools": true | false
    }
    只返回JSON，不要其他内容。"""
    
    response = await llm.ainvoke([
        SystemMessage(content=system_prompt),
        state["messages"][-1],
    ])
    
    import json
    routing_decision = json.loads(response.content)
    
    return {
        "route": routing_decision["domain"],
        "complexity": routing_decision["complexity"],
    }

def route_to_agent(state: RouterState) -> str:
    """根据路由决策选择Agent"""
    route = state.get("route", "general")
    complexity = state.get("complexity", "simple")
    
    if complexity == "simple":
        return "fast_agent"
    elif route == "technical":
        return "tech_expert"
    elif route == "creative":
        return "creative_agent"
    elif route == "analysis":
        return "analyst_agent"
    else:
        return "general_agent"

# 构建路由图
graph = StateGraph(RouterState)

graph.add_node("router", intelligent_router)
graph.add_node("fast_agent", FastAgent().run)
graph.add_node("tech_expert", TechExpert().run)
graph.add_node("creative_agent", CreativeAgent().run)
graph.add_node("analyst_agent", AnalystAgent().run)
graph.add_node("general_agent", GeneralAgent().run)

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_to_agent)

# 所有Agent都连接到END
for agent in ["fast_agent", "tech_expert", "creative_agent", "analyst_agent", "general_agent"]:
    graph.add_edge(agent, END)

4.2 多级路由与子图

复杂系统可以将子任务封装为子图：

def create_research_subgraph() -> StateGraph:
    """创建研究子图"""
    subgraph = StateGraph(ResearchState)
    
    subgraph.add_node("plan", plan_research)
    subgraph.add_node("search", search_sources)
    subgraph.add_node("synthesize", synthesize_findings)
    subgraph.add_node("verify", verify_claims)
    
    subgraph.add_edge(START, "plan")
    subgraph.add_edge("plan", "search")
    subgraph.add_edge("search", "synthesize")
    subgraph.add_edge("synthesize", "verify")
    
    def needs_more_research(state):
        if state.get("confidence", 0) < 0.8:
            return "search"
        return END
    
    subgraph.add_conditional_edges("verify", needs_more_research)
    
    return subgraph.compile()

# 在主图中使用子图
main_graph = StateGraph(MainState)

research_subgraph = create_research_subgraph()
main_graph.add_node("researcher", research_subgraph)
main_graph.add_node("writer", write_report)
main_graph.add_node("reviewer", review_report)

main_graph.add_edge(START, "researcher")
main_graph.add_edge("researcher", "writer")
main_graph.add_edge("writer", "reviewer")
main_graph.add_edge("reviewer", END)

5. 多Agent编排策略

5.1 监督者模式（Supervisor）

一个监督Agent负责协调其他Agent的工作：

from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage

class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str
    task_complete: bool
    agent_history: list[str]

class SupervisorAgent:
    """监督者Agent - 协调多个专业Agent"""
    
    def __init__(self, agents: dict[str, callable]):
        self.agents = agents
        self.system_prompt = f"""你是一个任务协调者。根据当前对话状态，决定下一步应该由哪个Agent处理。

可用的Agent:
{self._format_agent_descriptions()}

返回JSON: {{"next_agent": "agent_name", "reason": "选择原因"}}
如果任务已完成，返回: {{"next_agent": "FINISH", "reason": "完成原因"}}
"""
    
    def _format_agent_descriptions(self) -> str:
        descriptions = {
            "researcher": "负责信息检索和资料收集",
            "coder": "负责代码编写和技术实现",
            "writer": "负责文档撰写和内容创作",
            "reviewer": "负责质量审核和结果验证",
        }
        return "\n".join(f"- {name}: {desc}" for name, desc in descriptions.items())
    
    async def decide_next(self, state: SupervisorState) -> dict:
        """决定下一个执行的Agent"""
        from langchain_core.messages import SystemMessage
        
        response = await llm.ainvoke([
            SystemMessage(content=self.system_prompt),
            *state["messages"],
        ])
        
        import json
        decision = json.loads(response.content)
        
        return {
            "next_agent": decision["next_agent"],
            "agent_history": state.get("agent_history", []) + [decision["next_agent"]],
        }

def route_to_next(state: SupervisorState) -> str:
    """路由到下一个Agent"""
    next_agent = state.get("next_agent", "FINISH")
    if next_agent == "FINISH":
        return END
    return next_agent

# 构建监督者图
graph = StateGraph(SupervisorState)

supervisor = SupervisorAgent(agents={...})

graph.add_node("supervisor", supervisor.decide_next)
graph.add_node("researcher", researcher_agent.run)
graph.add_node("coder", coder_agent.run)
graph.add_node("writer", writer_agent.run)
graph.add_node("reviewer", reviewer_agent.run)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_to_next)

# 所有Agent完成后回到监督者
for agent_name in ["researcher", "coder", "writer", "reviewer"]:
    graph.add_edge(agent_name, "supervisor")

5.2 层级委派模式（Hierarchical Delegation）

上级Agent将任务委派给下级Agent，下级可以进一步委派：

class HierarchicalState(TypedDict):
    messages: Annotated[list, add_messages]
    task_queue: list[dict]
    completed_tasks: list[dict]
    delegation_level: int

class TeamLead:
    """团队负责人 - 任务分解与委派"""
    
    async def decompose_task(self, state: HierarchicalState) -> dict:
        """将复杂任务分解为子任务"""
        task = state["messages"][-1].content
        
        system_prompt = """将以下任务分解为2-5个可执行的子任务。
        返回JSON数组: [{"id": 1, "task": "描述", "assignee": "agent_name", "priority": 1-5}]"""
        
        response = await llm.ainvoke([
            SystemMessage(content=system_prompt),
            HumanMessage(content=task),
        ])
        
        import json
        subtasks = json.loads(response.content)
        
        return {
            "task_queue": subtasks,
            "delegation_level": state.get("delegation_level", 0) + 1,
        }
    
    async def review_results(self, state: HierarchicalState) -> dict:
        """审查子任务结果"""
        completed = state.get("completed_tasks", [])
        # 检查是否所有任务都完成
        queue = state.get("task_queue", [])
        all_done = len(completed) >= len(queue)
        
        if all_done:
            # 汇总结果
            summary = await self._summarize_results(completed)
            return {
                "messages": [AIMessage(content=summary)],
                "task_complete": True,
            }
        return {"task_complete": False}

5.3 辩论模式（Debate）

多个Agent就同一问题给出不同观点，通过辩论达成共识：

class DebateState(TypedDict):
    topic: str
    messages: Annotated[list, add_messages]
    positions: dict[str, str]      # agent_name -> 立场
    round_number: int
    consensus_reached: bool

class DebateOrchestrator:
    """辩论协调器"""
    
    MAX_ROUNDS = 3
    
    async def collect_positions(self, state: DebateState) -> dict:
        """收集各方立场"""
        positions = {}
        
        for agent_name, agent in self.agents.items():
            prompt = f"""关于以下话题，给出你的专业观点：
            话题: {state['topic']}
            
            {self._get_debate_context(state)}
            
            要求：
            1. 给出明确立场
            2. 提供论据支持
            3. 指出其他观点的潜在问题"""
            
            response = await agent.ainvoke(prompt)
            positions[agent_name] = response.content
        
        return {"positions": positions, "round_number": state.get("round_number", 0) + 1}
    
    async def synthesize(self, state: DebateState) -> dict:
        """综合各方观点，形成结论"""
        positions = state.get("positions", {})
        
        synthesis_prompt = f"""基于以下多方观点，形成一个平衡、全面的结论：
        
        {self._format_positions(positions)}
        
        要求：
        1. 总结各方的核心观点
        2. 指出共识点和分歧点
        3. 给出综合结论和建议"""
        
        response = await llm.ainvoke(synthesis_prompt)
        
        return {
            "messages": [AIMessage(content=response.content)],
            "consensus_reached": True,
        }

6. 人机协作节点

6.1 Human-in-the-Loop设计

LangGraph原生支持在图中插入人工审批节点：

from langgraph.types import interrupt, Command

class ApprovalState(TypedDict):
    messages: Annotated[list, add_messages]
    pending_action: dict
    approval_status: str
    human_feedback: str

def human_approval_node(state: ApprovalState) -> dict:
    """人工审批节点 - 暂停执行等待人工确认"""
    
    pending = state.get("pending_action", {})
    
    # interrupt() 会暂停图的执行，将控制权返回给调用者
    human_decision = interrupt({
        "type": "approval_request",
        "action": pending.get("description", ""),
        "risk_level": pending.get("risk_level", "unknown"),
        "details": pending,
        "options": ["approve", "reject", "modify"],
    })
    
    return {
        "approval_status": human_decision.get("decision", "reject"),
        "human_feedback": human_decision.get("feedback", ""),
    }

def after_approval(state: ApprovalState) -> str:
    """根据审批结果路由"""
    status = state.get("approval_status", "reject")
    if status == "approve":
        return "execute_action"
    elif status == "modify":
        return "modify_action"
    else:
        return "cancel_action"

# 构建含人工审批的图
graph = StateGraph(ApprovalState)

graph.add_node("prepare_action", prepare_action)
graph.add_node("human_approval", human_approval_node)
graph.add_node("execute_action", execute_action)
graph.add_node("modify_action", modify_action)
graph.add_node("cancel_action", cancel_action)

graph.add_edge(START, "prepare_action")
graph.add_edge("prepare_action", "human_approval")
graph.add_conditional_edges("human_approval", after_approval)
graph.add_edge("execute_action", END)
graph.add_edge("modify_action", "prepare_action")  # 修改后重新提交
graph.add_edge("cancel_action", END)

6.2 使用图执行时处理中断

from langgraph.types import Command

async def run_with_human_approval(app, input_data, thread_config):
    """运行带人工审批的工作流"""
    
    # 第一次运行 - 会停在interrupt节点
    result = await app.ainvoke(input_data, config=thread_config)
    
    # 检查是否有中断
    if "__interrupt__" in result:
        interrupt_info = result["__interrupt__"][0]
        print(f"需要人工审批: {interrupt_info.value}")
        
        # 模拟人工决策（实际中可以来自Web界面、Slack等）
        human_decision = {
            "decision": "approve",
            "feedback": "同意执行，请注意限制条件",
        }
        
        # 恢复执行
        result = await app.ainvoke(
            Command(resume=human_decision),
            config=thread_config,
        )
    
    return result

7. 持久化状态与检查点

7.1 检查点机制

LangGraph的检查点机制允许图的状态被持久化和恢复：

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.memory import MemorySaver

# 方式1: 内存检查点（开发/测试用）
checkpointer = MemorySaver()

# 方式2: PostgreSQL检查点（生产用）
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/langgraph"
)

# 编译图时绑定检查点
app = graph.compile(checkpointer=checkpointer)

# 使用thread_id标识一个会话
config = {"configurable": {"thread_id": "user-session-123"}}

# 执行 - 状态会自动保存
result1 = await app.ainvoke({"messages": [HumanMessage("你好")]}, config=config)

# 恢复执行 - 从上次的状态继续
result2 = await app.ainvoke({"messages": [HumanMessage("继续")]}, config=config)

7.2 时间旅行调试

async def time_travel_debug(app, thread_id: str):
    """时间旅行调试 - 查看历史状态"""
    config = {"configurable": {"thread_id": thread_id}}
    
    # 获取该线程的所有检查点
    checkpoints = []
    async for checkpoint in app.aget_state_history(config):
        checkpoints.append(checkpoint)
    
    print(f"共 {len(checkpoints)} 个检查点:")
    for i, cp in enumerate(checkpoints):
        print(f"  [{i}] Step: {cp.metadata.get('step', 'N/A')}, "
              f"Node: {cp.metadata.get('source', 'N/A')}")
    
    # 恢复到某个历史状态
    if len(checkpoints) > 2:
        target_checkpoint = checkpoints[-3]  # 倒数第3个
        await app.aupdate_state(config, target_checkpoint.values)
        print(f"已恢复到检查点 [{len(checkpoints)-3}]")

7.3 自定义检查点存储

from langgraph.checkpoint.base import BaseCheckpointSaver
import json
import redis

class RedisCheckpointer(BaseCheckpointSaver):
    """基于Redis的检查点存储"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        super().__init__()
        self.redis = redis.from_url(redis_url)
        self.ttl = 86400 * 7  # 7天过期
    
    async def aput(self, config, checkpoint, metadata, new_versions):
        """保存检查点"""
        thread_id = config["configurable"]["thread_id"]
        key = f"checkpoint:{thread_id}:{checkpoint.id}"
        
        data = {
            "checkpoint": self._serialize(checkpoint),
            "metadata": metadata,
            "versions": new_versions,
        }
        
        self.redis.setex(key, self.ttl, json.dumps(data))
        # 同时维护一个有序列表用于历史查询
        self.redis.zadd(
            f"checkpoint_history:{thread_id}",
            {checkpoint.id: checkpoint.ts}
        )
    
    async def aget(self, config):
        """获取最新检查点"""
        thread_id = config["configurable"]["thread_id"]
        # 获取最新的checkpoint ID
        latest = self.redis.zrevrange(
            f"checkpoint_history:{thread_id}", 0, 0
        )
        if not latest:
            return None
        
        key = f"checkpoint:{thread_id}:{latest[0].decode()}"
        data = json.loads(self.redis.get(key))
        return self._deserialize(data["checkpoint"])

8. 流式输出

8.1 流式输出的类型

LangGraph支持多种流式输出模式：

# 模式1: 流式输出每个节点的最终结果
async for event in app.astream(input_data, config=config):
    print(f"节点完成: {event}")

# 模式2: 流式输出Token（LLM的逐字输出）
async for event in app.astream_events(input_data, config=config, version="v2"):
    if event["event"] == "on_chat_model_stream":
        token = event["data"]["chunk"].content
        print(token, end="", flush=True)

# 模式3: 流式输出特定节点
async for event in app.astream(
    input_data, 
    config=config,
    stream_mode="updates",  # 只输出状态更新
):
    for node_name, node_output in event.items():
        print(f"[{node_name}]: {node_output}")

8.2 实时流式输出实现

from langchain_core.messages import AIMessageChunk
from typing import AsyncGenerator

class StreamingAgentGraph:
    """支持流式输出的Agent图"""
    
    def __init__(self, app):
        self.app = app
    
    async def stream_response(
        self, 
        user_input: str, 
        thread_id: str
    ) -> AsyncGenerator[str, None]:
        """流式输出响应"""
        config = {"configurable": {"thread_id": thread_id}}
        input_data = {"messages": [HumanMessage(content=user_input)]}
        
        current_node = None
        
        async for event in self.app.astream_events(
            input_data, config=config, version="v2"
        ):
            kind = event["event"]
            
            # 节点开始
            if kind == "on_chain_start":
                node_name = event.get("name", "")
                if node_name not in ("LangGraph", "__start__"):
                    current_node = node_name
                    yield f"\n\n🔄 [{node_name}] 开始处理...\n"
            
            # LLM Token流
            elif kind == "on_chat_model_stream":
                chunk = event["data"]["chunk"]
                if isinstance(chunk, AIMessageChunk) and chunk.content:
                    yield chunk.content
            
            # 工具调用
            elif kind == "on_tool_start":
                tool_name = event.get("name", "unknown")
                yield f"\n🔧 调用工具: {tool_name}..."
            
            elif kind == "on_tool_end":
                yield " ✓\n"
            
            # 节点完成
            elif kind == "on_chain_end":
                node_name = event.get("name", "")
                if node_name not in ("LangGraph", "__start__"):
                    yield f"\n✅ [{node_name}] 完成\n"

# 使用示例
graph_app = graph.compile(checkpointer=MemorySaver())
streamer = StreamingAgentGraph(graph_app)

async for token in streamer.stream_response("帮我写一个Python排序算法", "session-1"):
    print(token, end="", flush=True)

9. 与CrewAI/AutoGen对比

9.1 架构对比

9.2 同一任务的三种实现

任务：根据用户需求，搜索资料 → 撰写文章 → 审核发布

LangGraph实现：

from langgraph.graph import StateGraph, START, END

class ArticleState(TypedDict):
    topic: str
    research: str
    draft: str
    review_result: str
    messages: Annotated[list, add_messages]

graph = StateGraph(ArticleState)

graph.add_node("researcher", research_agent)
graph.add_node("writer", writer_agent)
graph.add_node("reviewer", reviewer_agent)

graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")

def should_publish(state):
    if state.get("review_result") == "approved":
        return END
    return "writer"  # 不合格则重写

graph.add_conditional_edges("reviewer", should_publish)

app = graph.compile()

CrewAI实现：

from crewai import Agent, Task, Crew

researcher = Agent(
    role="研究员",
    goal="收集关于{topic}的全面资料",
    backstory="你是一位资深研究员",
    tools=[search_tool, web_scraper],
)

writer = Agent(
    role="撰稿人",
    goal="基于研究资料撰写高质量文章",
    backstory="你是一位专业写手",
)

reviewer = Agent(
    role="编辑",
    goal="审核文章质量和准确性",
    backstory="你是一位严格的编辑",
)

research_task = Task(
    description="收集关于{topic}的资料",
    expected_output="结构化的研究资料",
    agent=researcher,
)

writing_task = Task(
    description="基于研究资料撰写文章",
    expected_output="一篇完整的文章",
    agent=writer,
)

review_task = Task(
    description="审核文章，给出修改建议或通过",
    expected_output="审核结果和建议",
    agent=reviewer,
)

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "AI安全"})

AutoGen实现：

from autogen import AssistantAgent, UserProxyAgent

researcher = AssistantAgent(
    name="Researcher",
    system_message="你是研究员，负责收集资料。",
    llm_config={"model": "gpt-4"},
)

writer = AssistantAgent(
    name="Writer",
    system_message="你是撰稿人，基于资料写文章。",
    llm_config={"model": "gpt-4"},
)

reviewer = AssistantAgent(
    name="Reviewer",
    system_message="你是编辑，审核文章质量。",
    llm_config={"model": "gpt-4"},
)

user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    code_execution_config=False,
)

# 建立群聊
from autogen import GroupChat, GroupChatManager

groupchat = GroupChat(
    agents=[user_proxy, researcher, writer, reviewer],
    messages=[],
    max_round=10,
)

manager = GroupChatManager(groupchat=groupchat, llm_config={"model": "gpt-4"})

user_proxy.initiate_chat(manager, message="写一篇关于AI安全的文章")

9.3 选型建议

选择LangGraph当：
├── 需要复杂的条件路由和分支逻辑
├── 需要精确控制执行流程
├── 需要持久化状态和恢复能力
├── 需要人机协作审批流程
├── 部署到生产环境，需要可靠性
└── 团队有LangChain生态经验

选择CrewAI当：
├── 需要快速搭建原型
├── 任务结构相对简单（线性为主）
├── 重视角色定义和自然语言描述
├── 团队对Agent概念较新
└── 不需要复杂的路由和状态管理

选择AutoGen当：
├── 研究导向，探索多Agent对话
├── 需要代码执行能力
├── 重视Agent间的自由对话
├── 需要灵活的人类参与模式
└── 微软生态集成需求

10. 生产部署

10.1 部署架构

┌─────────────────────────────────────────────────┐
│                   负载均衡器                      │
└───────────────────────┬─────────────────────────┘
                        │
         ┌──────────────┼──────────────┐
         │              │              │
    ┌────▼────┐   ┌────▼────┐   ┌────▼────┐
    │ API     │   │ API     │   │ API     │
    │ Server  │   │ Server  │   │ Server  │
    │ (FastAPI)│  │ (FastAPI)│  │ (FastAPI)│
    └────┬────┘   └────┬────┘   └────┬────┘
         │              │              │
         └──────────────┼──────────────┘
                        │
              ┌─────────┴─────────┐
              │                   │
         ┌────▼────┐       ┌────▼────┐
         │ Redis   │       │PostgreSQL│
         │ (缓存)  │       │(检查点)  │
         └─────────┘       └─────────┘

10.2 FastAPI部署示例

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import StreamingResponse
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import uuid
import json

app = FastAPI(title="LangGraph Agent Service")

# 初始化
checkpointer = None  # 在startup事件中初始化
graph_app = None

@app.on_event("startup")
async def startup():
    global checkpointer, graph_app
    checkpointer = AsyncPostgresSaver.from_conn_string(
        "postgresql://user:pass@localhost:5432/langgraph"
    )
    await checkpointer.setup()
    graph_app = graph.compile(checkpointer=checkpointer)

@app.post("/chat")
async def chat(request: ChatRequest):
    """非流式聊天接口"""
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    
    input_data = {"messages": [HumanMessage(content=request.message)]}
    result = await graph_app.ainvoke(input_data, config=config)
    
    return {
        "thread_id": thread_id,
        "response": result["messages"][-1].content,
    }

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """流式聊天接口（SSE）"""
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    
    input_data = {"messages": [HumanMessage(content=request.message)]}
    
    async def event_generator():
        async for event in graph_app.astream_events(
            input_data, config=config, version="v2"
        ):
            if event["event"] == "on_chat_model_stream":
                chunk = event["data"]["chunk"]
                if chunk.content:
                    yield f"data: {json.dumps({'token': chunk.content})}\n\n"
            elif event["event"] == "on_chain_end":
                node_name = event.get("name", "")
                yield f"data: {json.dumps({'node_complete': node_name})}\n\n"
        
        yield f"data: {json.dumps({'done': True, 'thread_id': thread_id})}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
    )

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    """WebSocket聊天接口"""
    await websocket.accept()
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    
    try:
        while True:
            data = await websocket.receive_json()
            message = data.get("message", "")
            
            input_data = {"messages": [HumanMessage(content=message)]}
            
            async for event in graph_app.astream_events(
                input_data, config=config, version="v2"
            ):
                if event["event"] == "on_chat_model_stream":
                    chunk = event["data"]["chunk"]
                    if chunk.content:
                        await websocket.send_json({
                            "type": "token",
                            "content": chunk.content,
                        })
            
            await websocket.send_json({"type": "done"})
            
    except WebSocketDisconnect:
        pass

10.3 监控与可观测性

import time
import logging
from functools import wraps
from dataclasses import dataclass, field

@dataclass
class AgentMetrics:
    """Agent性能指标"""
    total_requests: int = 0
    total_tokens: int = 0
    total_latency: float = 0.0
    error_count: int = 0
    node_execution_times: dict = field(default_factory=dict)
    
    def record_request(self, latency: float, tokens: int, nodes_executed: list):
        self.total_requests += 1
        self.total_tokens += tokens
        self.total_latency += latency
        for node in nodes_executed:
            self.node_execution_times[node] = (
                self.node_execution_times.get(node, 0) + 1
            )
    
    def get_summary(self) -> dict:
        return {
            "total_requests": self.total_requests,
            "avg_latency": (
                self.total_latency / self.total_requests 
                if self.total_requests > 0 else 0
            ),
            "total_tokens": self.total_tokens,
            "error_rate": (
                self.error_count / self.total_requests 
                if self.total_requests > 0 else 0
            ),
            "node_usage": self.node_execution_times,
        }

class InstrumentedGraph:
    """带监控的图包装器"""
    
    def __init__(self, app, metrics: AgentMetrics):
        self.app = app
        self.metrics = metrics
        self.logger = logging.getLogger("langgraph.metrics")
    
    async def ainvoke(self, input_data, config=None):
        start_time = time.time()
        nodes_executed = []
        
        try:
            # 使用stream_events来追踪节点执行
            final_result = None
            async for event in self.app.astream_events(
                input_data, config=config, version="v2"
            ):
                if event["event"] == "on_chain_start":
                    node_name = event.get("name", "")
                    if node_name and node_name not in ("LangGraph",):
                        nodes_executed.append(node_name)
                
                if event["event"] == "on_chain_end" and event.get("name") == "LangGraph":
                    final_result = event["data"].get("output")
            
            latency = time.time() - start_time
            
            # 估算token使用
            estimated_tokens = self._estimate_tokens(input_data, final_result)
            
            self.metrics.record_request(latency, estimated_tokens, nodes_executed)
            self.logger.info(
                f"Request completed: latency={latency:.2f}s, "
                f"tokens={estimated_tokens}, nodes={nodes_executed}"
            )
            
            return final_result
            
        except Exception as e:
            self.metrics.error_count += 1
            self.logger.error(f"Request failed: {e}")
            raise

11. 复杂工作流实战：智能客服系统

11.1 系统设计

用户消息
    │
    ▼
┌─────────────┐
│ 意图分类器   │
└──────┬──────┘
       │
       ├──→ 咨询类 → 知识库检索 → 生成回答 → 质量检查 → 输出
       │
       ├──→ 投诉类 → 情绪安抚 → 记录工单 → 升级判断
       │                                      │
       │                    自动处理 ◄────────┤
       │                                      │
       │                    人工转接 ◄────────┘
       │
       ├──→ 操作类 → 身份验证 → 权限检查 → 执行操作 → 结果确认
       │
       └──→ 闲聊类 → 闲聊Agent → 输出

11.2 完整实现

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.types import interrupt, Command
from langgraph.checkpoint.postgres import PostgresSaver
from datetime import datetime

# ============ 状态定义 ============
class CustomerServiceState(TypedDict):
    messages: Annotated[list, add_messages]
    customer_id: str
    intent: str                    # 咨询/投诉/操作/闲聊
    sentiment: str                 # positive/neutral/negative/angry
    knowledge_results: list        # 知识库检索结果
    ticket_id: str                 # 工单ID
    escalation_needed: bool        # 是否需要升级
    response: str                  # 最终回复
    metadata: dict                 # 元数据

# ============ 节点实现 ============

async def classify_intent(state: CustomerServiceState) -> dict:
    """意图分类 + 情感分析"""
    from langchain_core.messages import SystemMessage
    
    system_prompt = """分析用户消息，返回JSON:
    {
        "intent": "inquiry" | "complaint" | "operation" | "chitchat",
        "sentiment": "positive" | "neutral" | "negative" | "angry",
        "keywords": ["关键词1", "关键词2"],
        "urgency": "low" | "medium" | "high"
    }"""
    
    response = await llm.ainvoke([
        SystemMessage(content=system_prompt),
        state["messages"][-1],
    ])
    
    import json
    result = json.loads(response.content)
    
    return {
        "intent": result["intent"],
        "sentiment": result["sentiment"],
        "metadata": {
            **state.get("metadata", {}),
            "keywords": result.get("keywords", []),
            "urgency": result.get("urgency", "medium"),
            "classified_at": datetime.now().isoformat(),
        },
    }

async def search_knowledge_base(state: CustomerServiceState) -> dict:
    """检索知识库"""
    query = state["messages"][-1].content
    keywords = state.get("metadata", {}).get("keywords", [])
    
    # 向量检索 + 关键词检索
    results = await vector_store.asimilarity_search(
        query, k=5, filter={"category": "faq"}
    )
    
    return {
        "knowledge_results": [
            {"content": doc.page_content, "score": doc.metadata.get("score", 0)}
            for doc in results
        ],
    }

async def generate_response(state: CustomerServiceState) -> dict:
    """生成客服回复"""
    knowledge = state.get("knowledge_results", [])
    intent = state.get("intent", "inquiry")
    sentiment = state.get("sentiment", "neutral")
    
    knowledge_text = "\n".join([r["content"] for r in knowledge[:3]])
    
    system_prompt = f"""你是一位专业的客服代表。
    
    当前情况:
    - 用户意图: {intent}
    - 用户情绪: {sentiment}
    - 参考资料: {knowledge_text}
    
    回复要求:
    1. 语气友好专业
    2. 如果用户情绪负面，先表示理解
    3. 基于参考资料回答，不要编造
    4. 提供具体可操作的建议"""
    
    response = await llm.ainvoke([
        SystemMessage(content=system_prompt),
        *state["messages"],
    ])
    
    return {
        "response": response.content,
        "messages": [AIMessage(content=response.content)],
    }

async def handle_complaint(state: CustomerServiceState) -> dict:
    """处理投诉"""
    sentiment = state.get("sentiment", "neutral")
    
    # 严重投诉自动创建工单
    if sentiment in ("negative", "angry"):
        ticket_id = await create_ticket(
            customer_id=state.get("customer_id", "unknown"),
            category="complaint",
            description=state["messages"][-1].content,
            priority="high" if sentiment == "angry" else "medium",
        )
        return {
            "ticket_id": ticket_id,
            "escalation_needed": sentiment == "angry",
        }
    return {}

async def human_agent_node(state: CustomerServiceState) -> dict:
    """转接人工坐席"""
    # 使用interrupt暂停，等待人工坐席接管
    human_response = interrupt({
        "type": "escalation",
        "customer_id": state.get("customer_id"),
        "ticket_id": state.get("ticket_id"),
        "conversation_history": [
            msg.content for msg in state["messages"][-5:]
        ],
        "sentiment": state.get("sentiment"),
    })
    
    return {
        "messages": [AIMessage(content=human_response.get("response", ""))],
        "response": human_response.get("response", ""),
    }

async def execute_operation(state: CustomerServiceState) -> dict:
    """执行用户请求的操作（如查询订单、修改信息等）"""
    # 需要人工确认的敏感操作
    if state.get("metadata", {}).get("urgency") == "high":
        confirmation = interrupt({
            "type": "operation_confirmation",
            "operation": state["messages"][-1].content,
            "customer_id": state.get("customer_id"),
        })
        
        if not confirmation.get("approved"):
            return {
                "response": "操作已被取消。如有疑问请联系人工客服。",
                "messages": [AIMessage(content="操作已被取消。")],
            }
    
    # 执行操作
    result = await perform_operation(state)
    return {
        "response": result["message"],
        "messages": [AIMessage(content=result["message"])],
    }

# ============ 路由函数 ============

def route_by_intent(state: CustomerServiceState) -> str:
    """根据意图路由"""
    intent = state.get("intent", "chitchat")
    routing = {
        "inquiry": "knowledge_search",
        "complaint": "complaint_handler",
        "operation": "operation_executor",
        "chitchat": "responder",
    }
    return routing.get(intent, "responder")

def after_complaint(state: CustomerServiceState) -> str:
    """投诉处理后路由"""
    if state.get("escalation_needed"):
        return "human_agent"
    return "responder"

# ============ 构建图 ============

graph = StateGraph(CustomerServiceState)

# 注册节点
graph.add_node("classifier", classify_intent)
graph.add_node("knowledge_search", search_knowledge_base)
graph.add_node("complaint_handler", handle_complaint)
graph.add_node("operation_executor", execute_operation)
graph.add_node("responder", generate_response)
graph.add_node("human_agent", human_agent_node)

# 定义边
graph.add_edge(START, "classifier")
graph.add_conditional_edges("classifier", route_by_intent)
graph.add_edge("knowledge_search", "responder")
graph.add_conditional_edges("complaint_handler", after_complaint)
graph.add_edge("operation_executor", "responder")
graph.add_edge("responder", END)
graph.add_edge("human_agent", END)

# 编译
checkpointer = PostgresSaver.from_conn_string(DB_URL)
customer_service_app = graph.compile(checkpointer=checkpointer)

12. 复杂工作流实战：自动化研究助手

12.1 研究助手工作流

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    research_question: str
    sub_questions: list[str]
    search_results: list[dict]
    analysis: str
    report: str
    citations: list[str]
    quality_score: float
    iteration: int

async def decompose_question(state: ResearchState) -> dict:
    """将研究问题分解为子问题"""
    question = state.get("research_question") or state["messages"][-1].content
    
    prompt = f"""将以下研究问题分解为3-5个子问题，便于系统性研究:
    
    研究问题: {question}
    
    返回JSON数组: ["子问题1", "子问题2", ...]"""
    
    response = await llm.ainvoke(prompt)
    sub_questions = json.loads(response.content)
    
    return {
        "research_question": question,
        "sub_questions": sub_questions,
        "iteration": state.get("iteration", 0) + 1,
    }

async def parallel_search(state: ResearchState) -> dict:
    """并行搜索多个子问题"""
    import asyncio
    
    sub_questions = state.get("sub_questions", [])
    
    async def search_one(q: str) -> list[dict]:
        results = await web_search(q)
        return results
    
    all_results = await asyncio.gather(*[
        search_one(q) for q in sub_questions
    ])
    
    # 合并结果
    merged = []
    for results in all_results:
        merged.extend(results)
    
    return {"search_results": merged}

async def synthesize_analysis(state: ResearchState) -> dict:
    """综合分析搜索结果"""
    results = state.get("search_results", [])
    question = state.get("research_question", "")
    
    results_text = "\n\n".join([
        f"来源: {r.get('title', 'N/A')}\n内容: {r.get('snippet', '')}"
        for r in results[:20]
    ])
    
    prompt = f"""基于以下搜索结果，对研究问题进行深入分析:

    研究问题: {question}
    
    搜索结果:
    {results_text}
    
    要求:
    1. 综合各方信息，形成系统性分析
    2. 标注信息来源（使用[1], [2]等标注）
    3. 指出信息之间的关联和矛盾
    4. 提出需要进一步研究的问题"""
    
    response = await llm.ainvoke(prompt)
    
    return {
        "analysis": response.content,
        "citations": [r.get("url", "") for r in results[:20]],
    }

async def evaluate_quality(state: ResearchState) -> dict:
    """评估研究质量"""
    analysis = state.get("analysis", "")
    question = state.get("research_question", "")
    
    prompt = f"""评估以下研究分析的质量（0-1分）:
    
    研究问题: {question}
    分析内容: {analysis[:2000]}
    
    评估维度:
    1. 信息全面性
    2. 逻辑连贯性
    3. 来源可靠性
    4. 结论合理性
    
    返回JSON: {{"score": 0.85, "feedback": "改进建议"}}"""
    
    response = await llm.ainvoke(prompt)
    evaluation = json.loads(response.content)
    
    return {"quality_score": evaluation["score"]}

async def generate_report(state: ResearchState) -> dict:
    """生成最终研究报告"""
    analysis = state.get("analysis", "")
    question = state.get("research_question", "")
    citations = state.get("citations", [])
    
    prompt = f"""基于以下分析，生成一份结构化的研究报告:

    研究问题: {question}
    分析: {analysis}
    
    报告格式:
    # 研究报告: {question}
    
    ## 摘要
    ## 背景
    ## 主要发现
    ## 深入分析
    ## 结论与建议
    ## 参考文献"""
    
    response = await llm.ainvoke(prompt)
    
    return {
        "report": response.content,
        "messages": [AIMessage(content=response.content)],
    }

# 条件路由
def should_continue_research(state: ResearchState) -> str:
    score = state.get("quality_score", 0)
    iteration = state.get("iteration", 0)
    
    if score >= 0.8 or iteration >= 3:
        return "report_writer"
    return "question_decomposer"  # 质量不够，继续研究

# 构建研究助手图
graph = StateGraph(ResearchState)

graph.add_node("question_decomposer", decompose_question)
graph.add_node("web_searcher", parallel_search)
graph.add_node("analyzer", synthesize_analysis)
graph.add_node("evaluator", evaluate_quality)
graph.add_node("report_writer", generate_report)

graph.add_edge(START, "question_decomposer")
graph.add_edge("question_decomposer", "web_searcher")
graph.add_edge("web_searcher", "analyzer")
graph.add_edge("analyzer", "evaluator")
graph.add_conditional_edges("evaluator", should_continue_research)
graph.add_edge("report_writer", END)

research_app = graph.compile(checkpointer=MemorySaver())

13. 性能优化与调试

13.1 性能优化策略

# 策略1: 节点级缓存
from functools import lru_cache
import hashlib

class CachedNode:
    """带缓存的节点包装器"""
    
    def __init__(self, func, ttl: int = 3600):
        self.func = func
        self.ttl = ttl
        self.cache = {}
    
    async def __call__(self, state: dict) -> dict:
        # 基于输入状态生成缓存键
        cache_key = hashlib.md5(
            json.dumps(state, sort_keys=True, default=str).encode()
        ).hexdigest()
        
        if cache_key in self.cache:
            result, timestamp = self.cache[cache_key]
            if time.time() - timestamp < self.ttl:
                return result
        
        result = await self.func(state)
        self.cache[cache_key] = (result, time.time())
        return result

# 策略2: 并行执行独立节点
from langgraph.types import Send

def fan_out_to_parallel(state: ResearchState) -> list:
    """将搜索任务并行分发"""
    return [
        Send("searcher", {"query": q}) 
        for q in state.get("sub_questions", [])
    ]

# 策略3: Token使用优化
class TokenOptimizer:
    """Token使用优化器"""
    
    @staticmethod
    def compress_context(messages: list, max_tokens: int = 4000) -> list:
        """压缩上下文，保留关键信息"""
        if not messages:
            return messages
        
        # 保留系统消息
        system_msgs = [m for m in messages if m.type == "system"]
        other_msgs = [m for m in messages if m.type != "system"]
        
        # 保留最近的消息
        recent = other_msgs[-10:]
        
        # 如果还是太长，摘要化较早的消息
        total_tokens = sum(len(m.content) // 2 for m in system_msgs + recent)
        
        if total_tokens > max_tokens:
            # 只保留最后几条
            recent = recent[-5:]
        
        return system_msgs + recent

13.2 调试工具

class GraphDebugger:
    """图调试工具"""
    
    def __init__(self, app):
        self.app = app
        self.execution_trace = []
    
    async def debug_invoke(self, input_data, config=None):
        """带详细跟踪的执行"""
        async for event in self.app.astream_events(
            input_data, config=config, version="v2"
        ):
            kind = event["event"]
            name = event.get("name", "")
            
            if kind == "on_chain_start" and name not in ("LangGraph",):
                self.execution_trace.append({
                    "node": name,
                    "start_time": time.time(),
                    "input": event["data"].get("input"),
                })
                print(f"▶ [{name}] 开始")
            
            elif kind == "on_chain_end" and name not in ("LangGraph",):
                if self.execution_trace:
                    self.execution_trace[-1]["end_time"] = time.time()
                    self.execution_trace[-1]["output"] = event["data"].get("output")
                
                elapsed = self.execution_trace[-1]["end_time"] - self.execution_trace[-1]["start_time"]
                print(f"✅ [{name}] 完成 ({elapsed:.2f}s)")
            
            elif kind == "on_chat_model_start":
                print(f"  🤖 LLM调用: {name}")
            
            elif kind == "on_tool_start":
                print(f"  🔧 工具调用: {name}")
            
            elif kind == "on_chat_model_end":
                tokens = event["data"].get("output", {}).get("usage", {})
                if tokens:
                    print(f"  📊 Token: {tokens}")
        
        return self.execution_trace
    
    def print_trace(self):
        """打印执行轨迹"""
        print("\n" + "=" * 60)
        print("执行轨迹:")
        print("=" * 60)
        for i, step in enumerate(self.execution_trace):
            elapsed = step.get("end_time", 0) - step.get("start_time", 0)
            print(f"  [{i+1}] {step['node']} ({elapsed:.2f}s)")
        print("=" * 60)

14. 最佳实践与常见陷阱

14.1 最佳实践

状态设计

• 状态字段使用类型注解，确保类型安全
• 选择合适的reducer策略（追加 vs 覆盖 vs 合并）
• 避免在状态中存储过大的数据，使用引用代替
• 为状态字段提供合理的默认值

节点设计

• 节点函数应该是幂等的（相同输入产生相同输出）
• 单个节点职责单一，避免过大的节点
• 节点返回值只包含需要更新的状态字段
• 异步节点使用async/await提高并发性能

图结构

• 避免过深的嵌套子图（最多2-3层）
• 每个条件分支都要有明确的终止条件
• 循环路径必须有退出条件和最大迭代次数
• 使用有意义的节点名称，便于调试

生产部署

• 使用PostgreSQL等持久化存储作为检查点后端
• 实现完善的错误处理和重试机制
• 添加监控和告警（延迟、错误率、Token使用）
• 使用流式输出提升用户体验

14.2 常见陷阱

陷阱1：状态合并冲突

# ❌ 错误：多个节点同时更新同一字段
def node_a(state):
    return {"results": {"key_a": "value_a"}}

def node_b(state):
    return {"results": {"key_b": "value_b"}}

# 并行执行时，后执行的会覆盖先执行的

# ✅ 正确：使用reducer合并
class State(TypedDict):
    results: Annotated[dict, lambda a, b: {**a, **b}]

陷阱2：无限循环

# ❌ 错误：没有退出条件
def should_retry(state):
    return "retry"  # 永远重试

# ✅ 正确：添加退出条件
def should_retry(state):
    if state.get("retry_count", 0) >= 3:
        return "give_up"
    if state.get("quality_score", 0) >= 0.8:
        return "done"
    return "retry"

陷阱3：内存泄漏

# ❌ 错误：状态中存储大量数据
class BadState(TypedDict):
    all_documents: list  # 可能无限增长

# ✅ 正确：只存储引用
class GoodState(TypedDict):
    document_ids: list   # 存储ID，需要时再查询

陷阱4：忽略错误处理

# ❌ 错误：节点没有错误处理
async def risky_node(state):
    result = await external_api_call()  # 可能失败
    return {"result": result}

# ✅ 正确：添加错误处理
async def safe_node(state):
    try:
        result = await external_api_call()
        return {"result": result, "status": "success"}
    except Exception as e:
        return {
            "result": None,
            "status": "error",
            "error_message": str(e),
        }

总结

LangGraph为多Agent协作提供了一个强大、灵活的编排框架。通过本文的学习，你应该已经掌握：

1. 核心概念：状态图、节点、边、检查点
2. 设计模式：线性流水线、路由分发、循环重试、并行扇出
3. 高级特性：人机协作、持久化状态、流式输出
4. 生产部署：FastAPI集成、监控、性能优化
5. 框架选型：与CrewAI、AutoGen的对比和适用场景

建议的学习路径：

基础 → 线性图 → 条件路由 → 循环 → 人机协作
                                            ↓
生产 ← 监控优化 ← 部署 ← 持久化 ← 多Agent编排

LangGraph的生态还在快速发展中，建议关注官方文档和GitHub仓库，及时了解新特性和最佳实践的更新。