LangGraph实战：从0到1构建生产级AI智能体工作流

2025年,AI智能体(Agent)从概念验证走向生产环境,但现实残酷:只有15%的AI Agent项目成功进入生产环境。 本文将基于LangGraph 1.0,通过完整实战案例,分享如何构建可维护、可扩展的生产级Agent工作流。

---

为什么选择LangGraph?

在开始之前,让我们先看一个真实的痛点:

传统Agent开发的三大困境

# 问题代码示例:用传统LangChain构建复杂Agent
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

class ComplexAgent:
    def __init__(self):
        self.memory = ConversationBufferMemory()
        self.chain = ConversationChain(
            llm=llm,
            memory=self.memory
        )
    
    def process(self, user_input):
        # 痛点1:状态管理碎片化
        context = self.get_context(user_input)
        
        # 痛点2:流程控制复杂,嵌套if-else
        if self.needs_search(user_input):
            result = self.search_web(user_input)
            if self.needs_analysis(result):
                analysis = self.analyze(result)
                if analysis.requires_human_review():
                    return self.human_review(analysis)
            return result
        else:
            return self.chat(user_input)
    
    # 痛点3:扩展性差,新增功能需要大量修改
    def get_context(self, input): ...
    def needs_search(self, input): ...
    def needs_analysis(self, result): ...
    def human_review(self, analysis): ...

核心问题:

• ❌ 状态管理混乱:状态散落在变量、内存、数据库中
• ❌ 流程控制复杂:大量嵌套的if-else,难以维护
• ❌ 扩展性差:新增功能需要重构大量代码

LangGraph的核心优势

LangGraph 1.0在2025年10月正式发布,带来了革命性的改进:

# LangGraph解决方案:声明式图结构
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated

# 1. 类型安全的状态定义
class AgentState(TypedDict):
    messages: Annotated[list, "对话历史"]
    context: Annotated[dict, "上下文信息"]
    tool_calls: Annotated[list, "工具调用记录"]
    current_step: Annotated[str, "当前步骤"]

# 2. 声明式工作流定义
workflow = StateGraph(AgentState)

# 3. 清晰的节点定义
workflow.add_node("classifier", classify_intent)
workflow.add_node("searcher", search_web)
workflow.add_node("analyzer", analyze_result)
workflow.add_node("reviewer", human_review)
workflow.add_node("responder", generate_response)

# 4. 可视化的边连接
workflow.set_entry_point("classifier")
workflow.add_conditional_edges(
    "classifier",
    route_by_intent,
    {
        "search": "searcher",
        "chat": "responder"
    }
)
workflow.add_edge("searcher", "analyzer")
workflow.add_conditional_edges(
    "analyzer",
    check_review_needed,
    {
        "review": "reviewer",
        "respond": "responder"
    }
)
workflow.add_edge("reviewer", "responder")
workflow.add_edge("responder", END)

优势对比:

• ✅ 状态管理集中:所有状态在TypedDict中明确定义
• ✅ 流程控制清晰:图结构替代嵌套if-else
• ✅ 扩展性强:新增节点和边即可扩展功能
• ✅ 可视化调试:LangSmith提供完整工作流可视化

---

实战案例:构建智能客服Agent

让我们通过一个完整的实战案例,展示如何用LangGraph构建生产级Agent。

场景描述

某电商公司的智能客服系统需要处理:

1. 订单查询:查询订单状态、物流信息
2. 售后处理:退换货、投诉处理
3. 产品咨询:产品信息、推荐建议
4. 复杂问题:需要人工介入的复杂场景

完整代码实现

1. 状态定义

from typing import TypedDict, Annotated, List, Literal
from operator import add
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

class CustomerServiceState(TypedDict):
    # 消息历史
    messages: Annotated[List[BaseMessage], add]
    
    # 意图识别结果
    intent: Annotated[str, "用户意图"]
    confidence: Annotated[float, "置信度"]
    
    # 订单信息
    order_id: Annotated[str | None, "订单ID"]
    order_info: Annotated[dict | None, "订单详情"]
    
    # 工具调用记录
    tool_calls: Annotated[List[dict], add]
    
    # 是否需要人工介入
    needs_human: Annotated[bool, "需要人工"]
    
    # 当前步骤(用于调试)
    current_step: Annotated[str, "当前节点"]

2. 节点实现

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 初始化LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

# ========== 节点1: 意图分类 ==========
INTENT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个客服意图分类器。请分析用户输入,判断其意图。

可用的意图类别:
1. order_query - 查询订单
2. after_sale - 售后服务(退换货、投诉)
3. product_inquiry - 产品咨询
4. complex - 复杂问题,需要人工处理

输出格式:
intent: <意图类别>
confidence: <0-1之间的置信度>
reason: <分类理由>"""),
    ("human", "{user_input}")
])

def classify_intent(state: CustomerServiceState):
    """意图分类节点"""
    user_input = state["messages"][-1].content
    
    prompt = INTENT_PROMPT.format(user_input=user_input)
    response = llm.invoke(prompt)
    
    # 解析响应
    intent = extract_intent(response.content)
    confidence = extract_confidence(response.content)
    
    return {
        "intent": intent,
        "confidence": confidence,
        "current_step": "classifier",
        "messages": [AIMessage(content=f"已识别意图: {intent} (置信度: {confidence})")]
    }

# ========== 节点2: 订单查询 ==========
@tool
def query_order_status(order_id: str) -> str:
    """查询订单状态"""
    # 这里调用实际的订单系统API
    # 简化示例
    if order_id == "123456":
        return "订单已发货,预计3天内到达"
    else:
        return "订单未找到"

def process_order_query(state: CustomerServiceState):
    """订单查询处理节点"""
    messages = state["messages"]
    
    # 从对话中提取订单ID
    order_id = extract_order_id(messages)
    
    if not order_id:
        return {
            "messages": [AIMessage(content="请提供您的订单ID")],
            "current_step": "order_query"
        }
    
    # 查询订单
    tool = query_order_status
    result = tool.invoke({"order_id": order_id})
    
    return {
        "order_id": order_id,
        "order_info": {"status": result},
        "messages": [AIMessage(content=result)],
        "current_step": "order_query"
    }

# ========== 节点3: 售后处理 ==========
AFTER_SALE_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个售后处理专员。处理用户的退换货请求。

处理流程:
1. 确认订单信息
2. 了解退换货原因
3. 提供处理方案

如果情况复杂,标记needs_human为True"""),
    ("human", "{context}")
])

def handle_after_sale(state: CustomerServiceState):
    """售后处理节点"""
    context = "\n".join([msg.content for msg in state["messages"]])
    
    prompt = AFTER_SALE_PROMPT.format(context=context)
    response = llm.invoke(prompt)
    
    # 判断是否需要人工介入
    needs_human = check_complexity(response.content)
    
    return {
        "needs_human": needs_human,
        "messages": [AIMessage(content=response.content)],
        "current_step": "after_sale"
    }

# ========== 节点4: 产品咨询 ==========
@tool
def get_product_info(product_name: str) -> str:
    """获取产品信息"""
    # 调用产品知识库
    return f"{product_name}是一款优质产品,评分4.8/5.0"

def handle_product_inquiry(state: CustomerServiceState):
    """产品咨询节点"""
    messages = state["messages"]
    product_name = extract_product_name(messages)
    
    if not product_name:
        return {
            "messages": [AIMessage(content="请问您想了解哪款产品?")],
            "current_step": "product_inquiry"
        }
    
    # 获取产品信息
    tool = get_product_info
    result = tool.invoke({"product_name": product_name})
    
    # 生成推荐建议
    recommendation_prompt = ChatPromptTemplate.from_messages([
        ("system", "基于产品信息,给出专业的推荐建议"),
        ("human", result)
    ])
    
    rec = llm.invoke(recommendation_prompt)
    
    return {
        "messages": [AIMessage(content=f"{result}\n\n{rec.content}")],
        "current_step": "product_inquiry"
    }

# ========== 节点5: 人工介入 ==========
def escalate_to_human(state: CustomerServiceState):
    """人工介入节点"""
    return {
        "messages": [AIMessage(content="您的问题较为复杂,已为您转接人工客服,请稍候...")],
        "needs_human": True,
        "current_step": "human_escalation"
    }

# ========== 节点6: 生成响应 ==========
def generate_final_response(state: CustomerServiceState):
    """生成最终响应"""
    # 获取最新的AI响应
    latest_message = [msg for msg in state["messages"] if isinstance(msg, AIMessage)][-1]
    
    return {
        "current_step": "response"
    }

3. 路由逻辑

def route_by_intent(state: CustomerServiceState) -> Literal["order_query", "after_sale", "product_inquiry", "human_escalation"]:
    """根据意图路由到不同节点"""
    intent = state["intent"]
    confidence = state["confidence"]
    
    # 置信度低,直接转人工
    if confidence < 0.6:
        return "human_escalation"
    
    # 根据意图路由
    if intent == "order_query":
        return "order_query"
    elif intent == "after_sale":
        return "after_sale"
    elif intent == "product_inquiry":
        return "product_inquiry"
    else:  # complex
        return "human_escalation"

def check_escalation_needed(state: CustomerServiceState) -> Literal["response", "human_escalation"]:
    """检查是否需要人工介入"""
    if state.get("needs_human", False):
        return "human_escalation"
    return "response"

4. 构建工作流图

def create_customer_service_workflow():
    """创建客服工作流"""
    workflow = StateGraph(CustomerServiceState)
    
    # 添加所有节点
    workflow.add_node("classifier", classify_intent)
    workflow.add_node("order_query", process_order_query)
    workflow.add_node("after_sale", handle_after_sale)
    workflow.add_node("product_inquiry", handle_product_inquiry)
    workflow.add_node("human_escalation", escalate_to_human)
    workflow.add_node("response", generate_final_response)
    
    # 设置入口
    workflow.set_entry_point("classifier")
    
    # 添加条件边
    workflow.add_conditional_edges(
        "classifier",
        route_by_intent,
        {
            "order_query": "order_query",
            "after_sale": "after_sale",
            "product_inquiry": "product_inquiry",
            "human_escalation": "human_escalation"
        }
    )
    
    # 订单查询后直接响应
    workflow.add_edge("order_query", "response")
    
    # 产品咨询后直接响应
    workflow.add_edge("product_inquiry", "response")
    
    # 售后处理后检查是否需要人工
    workflow.add_conditional_edges(
        "after_sale",
        check_escalation_needed,
        {
            "response": "response",
            "human_escalation": "human_escalation"
        }
    )
    
    # 人工介入后响应
    workflow.add_edge("human_escalation", "response")
    
    # 结束
    workflow.add_edge("response", END)
    
    return workflow.compile()

5. 运行与测试

# 创建工作流
workflow = create_customer_service_workflow()

# 测试用例1: 订单查询
test_case_1 = {
    "messages": [HumanMessage(content="我的订单123456什么时候能到?")],
    "intent": "",
    "confidence": 0.0,
    "order_id": None,
    "order_info": None,
    "tool_calls": [],
    "needs_human": False,
    "current_step": ""
}

result_1 = workflow.invoke(test_case_1)
print(result_1["messages"][-1].content)
# 输出: "订单已发货,预计3天内到达"

# 测试用例2: 售后处理
test_case_2 = {
    "messages": [HumanMessage(content="我买的衣服不合适想退货")],
    "intent": "",
    "confidence": 0.0,
    "order_id": None,
    "order_info": None,
    "tool_calls": [],
    "needs_human": False,
    "current_step": ""
}

result_2 = workflow.invoke(test_case_2)
print(result_2["messages"][-1].content)
# 输出: "您的订单编号是多少?我来帮您处理退货..."

# 测试用例3: 复杂问题
test_case_3 = {
    "messages": [HumanMessage(content="我要投诉你们的服务太差了,而且产品有问题,要求三倍赔偿!")],
    "intent": "",
    "confidence": 0.0,
    "order_id": None,
    "order_info": None,
    "tool_calls": [],
    "needs_human": False,
    "current_step": ""
}

result_3 = workflow.invoke(test_case_3)
print(result_3["messages"][-1].content)
# 输出: "您的问题较为复杂,已为您转接人工客服,请稍候..."

---

生产级部署的关键实践

上面的代码实现了核心逻辑,但要部署到生产环境,还需要考虑以下关键点:

1. 可观测性

from langsmith import LangSmith

# 启用LangSmith追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# 工作流编译时启用调试
workflow = create_customer_service_workflow().compile(
    debug=True,  # 详细日志
    interrupt_before=["human_escalation"]  # 在人工介入前暂停
)

可观测性指标:

• 每个节点的执行时间
• LLM调用次数和成本
• 意图分类准确率
• 人工介入率

2. 错误处理与重试

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def safe_llm_call(prompt):
    """带重试的LLM调用"""
    try:
        return llm.invoke(prompt)
    except Exception as e:
        logger.error(f"LLM调用失败: {e}")
        raise

def classify_intent_with_fallback(state: CustomerServiceState):
    """带降级的意图分类"""
    try:
        return classify_intent(state)
    except Exception as e:
        logger.warning(f"意图分类失败,使用默认分类: {e}")
        return {
            "intent": "complex",
            "confidence": 0.3,
            "current_step": "classifier_fallback"
        }

3. 性能优化

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# 启用缓存
set_llm_cache(InMemoryCache())

# 并行节点执行
from langgraph.graph import StateGraph, START, END

# 使用并发分支
def create_parallel_workflow():
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("task1", process_task1)
    workflow.add_node("task2", process_task2)
    workflow.add_node("task3", process_task3)
    workflow.add_node("aggregator", aggregate_results)
    
    # 并行执行
    workflow.add_edge(START, "task1")
    workflow.add_edge(START, "task2")
    workflow.add_edge(START, "task3")
    
    # 聚合结果
    workflow.add_edge("task1", "aggregator")
    workflow.add_edge("task2", "aggregator")
    workflow.add_edge("task3", "aggregator")
    workflow.add_edge("aggregator", END)
    
    return workflow.compile()

4. 状态持久化

from langgraph.checkpoint.memory import MemorySaver

# 使用内存检查点(开发环境)
memory = MemorySaver()

# 生产环境使用PostgreSQL或Redis
from langgraph.checkpoint.postgres import PostgresSaver

# PostgreSQL检查点
postgres_saver = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/agent_db"
)

# 编译工作流时启用检查点
workflow = create_customer_service_workflow().compile(
    checkpointer=postgres_saver
)

# 线程ID用于区分不同会话
thread_id = "user_session_123"
config = {"configurable": {"thread_id": thread_id}}

# 恢复状态
result = workflow.invoke(initial_state, config)

5. 监控与告警

import time
from prometheus_client import Counter, Histogram

# 定义监控指标
agent_requests = Counter('agent_requests_total', 'Total agent requests', ['intent'])
agent_duration = Histogram('agent_duration_seconds', 'Agent processing duration', ['node'])
llm_calls = Counter('llm_calls_total', 'Total LLM calls', ['model'])

def monitor_node(func):
    """节点监控装饰器"""
    def wrapper(state):
        start_time = time.time()
        node_name = func.__name__
        
        try:
            result = func(state)
            duration = time.time() - start_time
            agent_duration.labels(node=node_name).observe(duration)
            
            # 记录指标
            if "intent" in result:
                agent_requests.labels(intent=result["intent"]).inc()
            
            return result
        except Exception as e:
            logger.error(f"节点 {node_name} 执行失败: {e}")
            raise
    return wrapper

# 应用装饰器
@monitor_node
def classify_intent(state: CustomerServiceState):
    # ... 原有逻辑
    pass

---

真实生产案例分享

案例背景

某金融科技公司使用LangGraph重构其智能投研系统,替换原有的复杂if-else逻辑。

遇到的挑战

1. 性能问题:单次处理时间从2秒降低到800ms
2. 可维护性:代码行数减少60%
3. 扩展性:新增功能只需3天(原来需要2周)

架构设计

# 投研工作流核心架构
class ResearchState(TypedDict):
    query: str
    research_context: dict
    analysis_result: dict
    risk_assessment: dict
    investment_advice: str
    
workflow = StateGraph(ResearchState)

# 节点定义
workflow.add_node("query_parser", parse_research_query)
workflow.add_node("data_collector", collect_market_data)
workflow.add_node("analyst", analyze_data)
workflow.add_node("risk_assessor", assess_risk)
workflow.add_node("advisor", generate_advice)

# 工作流设计
workflow.set_entry_point("query_parser")
workflow.add_edge("query_parser", "data_collector")
workflow.add_edge("data_collector", "analyst")
workflow.add_edge("analyst", "risk_assessor")
workflow.add_conditional_edges(
    "risk_assessor",
    check_risk_level,
    {
        "low_risk": "advisor",
        "high_risk": "advisor"  # 高风险仍提供建议但带警告
    }
)
workflow.add_edge("advisor", END)

实际效果

指标	重构前	重构后	改进
平均响应时间	2000ms	800ms	60%↓
代码行数	5000+	2000	60%↓
新功能开发周期	2周	3天	78%↓
维护成本	高	低	-
团队上手时间	1个月	1周	75%↓

---

常见问题与解决方案

Q1: 如何处理长对话记忆?

问题:LangGraph的状态会无限增长,导致性能下降

解决方案:

def summarize_messages(state: CustomerServiceState):
    """消息总结节点"""
    messages = state["messages"]
    
    # 只保留最近10条消息
    recent_messages = messages[-10:]
    
    # 总结更早的消息
    if len(messages) > 10:
        old_messages = messages[:-10]
        summary_prompt = f"总结以下对话历史:\n{old_messages}"
        summary = llm.invoke(summary_prompt)
        recent_messages.insert(0, AIMessage(content=f"[历史总结]: {summary.content}"))
    
    return {"messages": recent_messages}

# 在工作流中定期触发总结
workflow.add_node("summarizer", summarize_messages)
workflow.add_edge("summarizer", END)

# 或者使用条件边,当消息过多时自动总结
workflow.add_conditional_edges(
    "classifier",
    lambda state: "summarizer" if len(state["messages"]) > 20 else "order_query",
    {"summarizer": "summarizer", "order_query": "order_query"}
)

Q2: 如何实现人工介入后的状态恢复?

问题:人工处理后,如何恢复到工作流继续执行?

解决方案:

# 在人工介入时保存状态
def escalate_to_human(state: CustomerServiceState):
    # 保存当前状态到数据库
    save_state_to_db(state["thread_id"], state)
    
    return {
        "messages": [AIMessage(content="已转接人工客服")],
        "needs_human": True,
        "current_step": "human_escalation"
    }

# 人工处理后的回调
def human_callback(thread_id: str, human_response: str):
    """人工处理后的回调"""
    # 从数据库恢复状态
    state = load_state_from_db(thread_id)
    
    # 添加人工响应
    state["messages"].append(HumanMessage(content=human_response))
    state["needs_human"] = False
    
    # 继续工作流
    return workflow.invoke(state)

Q3: 如何实现多Agent协作?

问题:复杂任务需要多个专业Agent协作完成

解决方案:

from langgraph.graph import StateGraph, END

class MultiAgentState(TypedDict):
    task: str
    research_result: dict
    design_result: dict
    development_result: dict
    testing_result: dict
    final_output: str

# Agent 1: 研究Agent
def research_agent(state: MultiAgentState):
    """研究Agent:负责信息收集"""
    # ... 研究逻辑
    return {"research_result": {...}}

# Agent 2: 设计Agent
def design_agent(state: MultiAgentState):
    """设计Agent:负责方案设计"""
    # 基于研究结果进行设计
    return {"design_result": {...}}

# Agent 3: 开发Agent
def development_agent(state: MultiAgentState):
    """开发Agent:负责代码实现"""
    # 基于设计方案进行开发
    return {"development_result": {...}}

# Agent 4: 测试Agent
def testing_agent(state: MultiAgentState):
    """测试Agent:负责质量保证"""
    # 测试开发结果
    return {"testing_result": {...}}

# 创建多Agent工作流
def create_multi_agent_workflow():
    workflow = StateGraph(MultiAgentState)
    
    # 添加Agent节点
    workflow.add_node("researcher", research_agent)
    workflow.add_node("designer", design_agent)
    workflow.add_node("developer", development_agent)
    workflow.add_node("tester", testing_agent)
    workflow.add_node("integrator", integrate_results)
    
    # 顺序执行
    workflow.set_entry_point("researcher")
    workflow.add_edge("researcher", "designer")
    workflow.add_edge("designer", "developer")
    workflow.add_edge("developer", "tester")
    workflow.add_edge("tester", "integrator")
    workflow.add_edge("integrator", END)
    
    return workflow.compile()

Q4: 如何实现动态工作流?

问题:根据运行时条件动态调整工作流结构

解决方案:

def dynamic_router(state: AgentState):
    """动态路由:根据状态动态选择下一个节点"""
    complexity = assess_task_complexity(state)
    
    if complexity == "low":
        return ["fast_path"]
    elif complexity == "medium":
        return ["standard_path"]
    else:
        return ["deep_analysis_path", "human_review"]

# 动态添加边
def create_dynamic_workflow():
    workflow = StateGraph(AgentState)
    
    # 基础节点
    workflow.add_node("analyzer", analyze_task)
    workflow.add_node("fast_path", simple_process)
    workflow.add_node("standard_path", normal_process)
    workflow.add_node("deep_analysis_path", deep_analysis)
    workflow.add_node("human_review", review_by_human)
    
    workflow.set_entry_point("analyzer")
    
    # 动态边:运行时决定路由
    workflow.add_conditional_edges(
        "analyzer",
        dynamic_router,
        {
            "fast_path": "fast_path",
            "standard_path": "standard_path",
            "deep_analysis_path": "deep_analysis_path"
        }
    )
    
    # 深度分析后可能需要人工审查
    workflow.add_conditional_edges(
        "deep_analysis_path",
        lambda s: "human_review" if s["risk_level"] == "high" else "response",
        {"human_review": "human_review", "response": "response"}
    )
    
    return workflow.compile()

---

性能优化最佳实践

1. 减少LLM调用次数

# ❌ 不好的实践:每个节点都调用LLM
def node1(state):
    response = llm.invoke(...)
    return {...}

def node2(state):
    response = llm.invoke(...)  # 又一次调用
    return {...}

# ✅ 好的实践:批量处理
def batch_process(state):
    # 一次性处理多个任务
    prompt = ChatPromptTemplate.from_messages([
        ("system", "同时完成以下任务:\n1. 意图识别\n2. 实体提取\n3. 情感分析"),
        ("human", "{user_input}")
    ])
    
    response = llm.invoke(prompt)
    
    # 解析多个结果
    intent = extract_intent(response.content)
    entities = extract_entities(response.content)
    sentiment = extract_sentiment(response.content)
    
    return {
        "intent": intent,
        "entities": entities,
        "sentiment": sentiment
    }

2. 使用流式输出

def streaming_response(state):
    """流式输出节点"""
    user_input = state["messages"][-1].content
    
    # 使用stream
    for chunk in llm.stream(user_input):
        yield chunk.content  # 实时返回结果

# 工作流中使用
workflow = create_workflow().compile()
for event in workflow.stream(initial_state):
    print(event)  # 实时打印中间结果

3. 并行化独立节点

from langgraph.graph import StateGraph, START, END

# 并行执行多个独立任务
def create_parallel_workflow():
    workflow = StateGraph(AgentState)
    
    workflow.add_node("task1", process_task1)
    workflow.add_node("task2", process_task2)
    workflow.add_node("task3", process_task3)
    workflow.add_node("aggregator", aggregate)
    
    # 所有任务从START开始
    workflow.add_edge(START, "task1")
    workflow.add_edge(START, "task2")
    workflow.add_edge(START, "task3")
    
    # 所有任务完成后聚合
    workflow.add_edge("task1", "aggregator")
    workflow.add_edge("task2", "aggregator")
    workflow.add_edge("task3", "aggregator")
    
    workflow.add_edge("aggregator", END)
    
    return workflow.compile()

4. 使用检查点实现断点续传

from langgraph.checkpoint.postgres import PostgresSaver

# 配置PostgreSQL检查点
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/agent_db"
)

workflow = create_workflow().compile(checkpointer=checkpointer)

# 长时间任务可以中断后恢复
config = {"configurable": {"thread_id": "long_task_123"}}

# 第一次执行
try:
    result = workflow.invoke(initial_state, config)
except KeyboardInterrupt:
    print("任务已暂停")

# 稍后恢复
result = workflow.invoke(None, config)  # 从检查点恢复

---

总结与展望

核心要点回顾

1. LangGraph的核心价值

   • 声明式图结构,替代复杂if-else
   • 集中式状态管理
   • 可视化调试与监控
   • 强大的扩展性

2. 生产级部署关键

   • 可观测性(LangSmith集成)
   • 错误处理与重试
   • 性能优化(缓存、并行)
   • 状态持久化
   • 监控告警

3. 实战经验

   • 从简单场景开始,逐步复杂化
   • 充分利用LangSmith调试
   • 建立完整的监控体系
   • 持续优化和迭代

未来展望

LangGraph 2.0值得期待:

• 更好的分布式支持
• 原生的多租户支持
• 更强的类型安全
• 性能进一步提升

行业趋势:

• AI Agent将成为标准基础设施
• 多Agent协作成为主流
• 人机协同模式成熟
• 企业级应用大规模落地

给开发者的建议

1. 立即行动:从今天开始尝试LangGraph
2. 实践为主:多写代码,少看理论
3. 社区参与:加入LangChain Discord,分享经验
4. 持续学习:技术迭代快,保持学习节奏

---

参考资源

• LangGraph官方文档
• LangSmith平台
• LangChain GitHub
• 实战案例集合

如果这篇文章对你有帮助,欢迎点赞、收藏和转发!有任何问题欢迎在评论区讨论。