开源智能体框架LangGraph深度实践：构建生产级Agent系统

LangGraph：从原型到生产的Agent框架

LangGraph最大的优势不在于功能丰富，而在于它对生产环境的认真对待——状态管理、检查点、人机协作、错误处理，这些生产级需求被设计在框架核心而非附加功能。

状态管理

定义Agent状态

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]  # 消息列表（追加）
    current_task: str                         # 当前任务
    completed_steps: List[str]               # 已完成步骤
    tool_results: dict                        # 工具结果
    error_count: int                          # 错误计数
    human_feedback: str                       # 人类反馈
    next_action: str                          # 下一步行动

# 创建图
graph = StateGraph(AgentState)

状态更新模式

def research_node(state: AgentState):
    """研究节点：执行信息检索"""
    query = state["current_task"]
    results = search_tool(query)
    
    # 状态更新（自动合并）
    return {
        "messages": [{"role": "assistant", "content": f"找到{len(results)}条结果"}],
        "tool_results": {"search": results},
        "completed_steps": state["completed_steps"] + ["research"],
        "next_action": "analyze"
    }

def analyze_node(state: AgentState):
    """分析节点：分析检索结果"""
    results = state["tool_results"]["search"]
    analysis = llm.analyze(results)
    
    return {
        "messages": [{"role": "assistant", "content": analysis}],
        "completed_steps": state["completed_steps"] + ["analyze"],
        "next_action": "write" if analysis else "research"  # 分析不足则重新检索
    }

检查点与恢复

持久化执行状态

from langgraph.checkpoint import MemorySaver, SqliteSaver

# 使用SQLite持久化
checkpointer = SqliteSaver.from_conn_string("agent.db")

graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_node("write", write_node)

graph.add_edge("research", "analyze")
graph.add_conditional_edges("analyze", lambda s: s["next_action"])
graph.add_edge("write", END)

app = graph.compile(checkpointer=checkpointer)

# 执行（可以中断和恢复）
config = {"configurable": {"thread_id": "task-123"}}
result = app.invoke(
    {"current_task": "分析AI芯片市场", "messages": []},
    config=config
)

# 恢复执行
restored = app.get_state(config)
# 可以从任意检查点恢复

检查点策略

class CheckpointStrategy:
    def __init__(self):
        self.saver = SqliteSaver.from_conn_string("checkpoints.db")
    
    def should_checkpoint(self, state):
        """决定是否需要检查点"""
        # 关键步骤后检查
        if state.get("completed_steps"):
            last_step = state["completed_steps"][-1]
            if last_step in ["research", "analyze", "write"]:
                return True
        
        # 错误后检查
        if state.get("error_count", 0) > 0:
            return True
        
        return False

人机协作

人工审批节点

# 在关键步骤前暂停，等待人工确认
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["publish"]  # 发布前暂停
)

# 执行到publish节点前会暂停
result = app.invoke(
    {"current_task": "撰写技术报告"},
    config={"configurable": {"thread_id": "task-456"}}
)

# 人工审查后继续
if human_approved:
    result = app.invoke(None, config=config)  # 传入None继续执行
else:
    # 人工提供修改意见
    result = app.invoke(
        {"human_feedback": "需要增加市场分析部分"},
        config=config
    )

交互式Agent

def human_interaction_node(state: AgentState):
    """需要人工输入的节点"""
    # 展示当前状态
    print(f"已完成步骤: {state['completed_steps']}")
    print(f"当前结果: {state.get('tool_results', {})}")
    
    # 请求人工输入
    feedback = input("请提供反馈（直接回车确认）: ")
    
    return {
        "human_feedback": feedback,
        "next_action": "revise" if feedback else "continue"
    }

错误处理与重试

节点级错误处理

def robust_node(state: AgentState, max_retries=3):
    """带错误处理的节点"""
    try:
        result = execute_task(state["current_task"])
        return {
            "tool_results": result,
            "error_count": 0,
            "next_action": "next"
        }
    except Exception as e:
        retry_count = state.get("error_count", 0) + 1
        
        if retry_count < max_retries:
            # 重试
            return {
                "error_count": retry_count,
                "next_action": "retry"  # 重新执行当前节点
            }
        else:
            # 超过重试次数，降级处理
            return {
                "error_count": 0,
                "messages": [{"role": "system", "content": f"任务失败: {e}"}],
                "next_action": "fallback"
            }

条件边实现重试逻辑

graph.add_node("execute", robust_node)
graph.add_node("fallback", fallback_node)

# 正常流程
graph.add_edge("execute", "next_node")

# 重试逻辑
graph.add_conditional_edges(
    "execute",
    lambda state: state.get("next_action"),
    {
        "retry": "execute",      # 重试当前节点
        "next": "next_node",     # 正常进入下一步
        "fallback": "fallback"   # 降级处理
    }
)

子图与模块化

# 将复杂Agent拆分为子图
def build_research_subgraph():
    """研究子图"""
    subgraph = StateGraph(ResearchState)
    subgraph.add_node("search", search_node)
    subgraph.add_node("filter", filter_node)
    subgraph.add_node("summarize", summarize_node)
    
    subgraph.add_edge("search", "filter")
    subgraph.add_edge("filter", "summarize")
    subgraph.add_edge("summarize", END)
    
    return subgraph.compile()

# 主图中嵌入子图
main_graph = StateGraph(AgentState)
main_graph.add_node("research", build_research_subgraph())  # 嵌入子图
main_graph.add_node("write", write_node)
main_graph.add_edge("research", "write")

并行执行

from langgraph.graph import StateGraph, END
import operator
from typing import Annotated

class ParallelState(TypedDict):
    task: str
    results: Annotated[list, operator.add]  # 并行结果追加

def parallel_research(state):
    """并行执行多个研究任务"""
    sub_tasks = decompose(state["task"])
    
    # 并行执行
    results = []
    for sub_task in sub_tasks:
        result = research_agent.run(sub_task)
        results.append(result)
    
    return {"results": results}

# 或者使用LangGraph的Send API实现真正的并行
from langgraph.constants import Send

def fan_out(state):
    """扇出并行任务"""
    sub_tasks = decompose(state["task"])
    return [
        Send("research_node", {"sub_task": st})
        for st in sub_tasks
    ]

生产部署

部署架构

class LangGraphDeployment:
    def __init__(self):
        self.config = {
            "runtime": {
                "framework": "FastAPI",
                "workers": 4,
                "timeout": 300,  # 5分钟超时
            },
            "checkpoint": {
                "backend": "PostgreSQL",  # 生产用PostgreSQL
                "cleanup_interval": 3600,  # 1小时清理一次
                "retention_days": 7,       # 保留7天
            },
            "monitoring": {
                "trace_enabled": True,
                "metrics": ["latency", "success_rate", "token_usage"],
                "alerting": {
                    "error_rate_threshold": 0.05,
                    "latency_p99_threshold": 30000,  # 30秒
                }
            }
        }
    
    def deploy(self):
        # FastAPI服务
        from fastapi import FastAPI
        app = FastAPI()
        
        @app.post("/agent/run")
        async def run_agent(task: str, thread_id: str):
            config = {"configurable": {"thread_id": thread_id}}
            result = await self.agent.ainvoke(
                {"current_task": task},
                config=config
            )
            return result
        
        return app

性能优化

class PerformanceOptimizer:
    def optimize_graph(self, graph):
        """图优化"""
        # 1. 节点合并：将总是顺序执行的节点合并
        # 2. 冗余边移除：移除不会被执行的边
        # 3. 缓存：对确定性节点启用缓存
        
        optimized = graph
        
        # 启用缓存
        for node in graph.nodes:
            if is_deterministic(node):
                node.enable_cache = True
                node.cache_ttl = 3600
        
        return optimized

监控与可观测性

class AgentMonitor:
    def __init__(self):
        self.traces = []
    
    def trace_execution(self, graph, input_state):
        """追踪Agent执行"""
        trace = {
            "input": input_state,
            "nodes_executed": [],
            "total_duration": 0,
            "token_usage": 0,
            "errors": []
        }
        
        for node_name, node_output in graph.stream(input_state):
            trace["nodes_executed"].append({
                "node": node_name,
                "duration": measure_duration(),
                "output": node_output,
                "timestamp": datetime.now()
            })
        
        return trace
    
    def visualize(self, trace):
        """可视化执行轨迹"""
        return {
            "graph": render_execution_graph(trace),
            "timeline": render_timeline(trace),
            "bottlenecks": identify_bottlenecks(trace)
        }

结语

LangGraph的设计哲学是"为生产而构建"。它的图模型提供了精确的控制力，检查点机制保障了可靠性，人机协作支持了复杂业务流程。对于需要从原型走向生产的Agent系统，LangGraph是最稳妥的选择。学习曲线确实陡峭，但这是为生产级功能付出的合理代价——在生产环境中，可靠性和可控性远比开发便利性重要。

LangGraph：从原型到生产的Agent框架#

状态管理#

定义Agent状态#

状态更新模式#

检查点与恢复#

持久化执行状态#

检查点策略#

人机协作#

人工审批节点#

交互式Agent#

错误处理与重试#

节点级错误处理#

条件边实现重试逻辑#

子图与模块化#

并行执行#

生产部署#

部署架构#

性能优化#

监控与可观测性#

结语#

📌 相关推荐