LangChain 和 LangGraph 是两个强大的人工智能开发框架。
一、为什么需要 LangGraph？
LangChain 虽然提供了丰富的工具链和 Agent 能力，但在实际企业级应用中面临一些挑战:
- 复杂状态管理困难: 传统 Chain 难以维护跨多个步骤的状态
- 循环流程不支持: 无法自然表达人类交互中的"重试 - 修正"过程
- 调试复杂度高: 执行流不透明，问题定位困难
- 长期任务控制弱: 缺乏对执行流的精确干预能力
LangGraph 通过图结构解决了这些问题，让你能够构建可预测、可控的智能体系统。
二、核心概念：图 vs 链
2.1 语言模型的三种执行模式
# 1. Prompt → LLM (最基础)
response = llm(prompt)
# 2. Chain (串行流程)
# question -> retriever -> prompt -> LLM -> answer
# 3. Graph (图结构)
# 节点之间可以有任意连接关系（包括循环）
2.2 关键术语
- Node: 计算单元，如
retrieval_node、llm_node、tool_node - Edge: 节点之间的连接，定义执行流方向
- State: 图的共享状态，所有节点读写同一份状态
- Cycle: 循环边，允许流程回到之前的节点
三、实战示例：状态管理的艺术
3.1 定义状态 (Schema)
这是 LangGraph 的核心！你的应用所有数据都存在于这个状态中:
from typing import TypedDict, Annotated, List
import operator
# 自定义状态类型
class AgentState(TypedDict):
"""Agent 的工作状态"""
# 用户输入
messages: Annotated[List, operator.add] # 累积消息历史
# 检索结果
retrieved_docs: List[str] # RAG 召回文档
# 工具调用参数
tools_called: List[str] # 已调用的工具列表
# 最终答案
final_answer: str # 结束标志
# 错误计数
error_count: int # 错误次数限制
关键点: Annotated[List, operator.add] 表示每次更新时追加而不是覆盖!
四、完整的 RAG Agent 实战
4.1 构建检索器
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
# 工具定义
@tool
def search_knowledge(query: str) -> str:
"""搜索知识库"""
# 这里调用向量数据库
return f"找到相关文档:{query}"
@tool
def run_code(code: str) -> str:
"""运行 Python 代码"""
# 安全执行代码的逻辑
result = eval(code) # 生产环境需加沙箱
return f"运行结果:{result}"
# 创建工具节点
tools = [search_knowledge, run_code]
tool_node = ToolNode(tools)
4.2 定义节点函数
每个节点都是一个普通 Python 函数:
from langchain_core.messages import HumanMessage
# 检索节点
def retrieval_node(state: AgentState):
"""从知识库检索相关信息"""
query = state['messages'][-1].content
docs = vector_db.similarity_search(query, k=3)
return {"retrieved_docs": [doc.page_content for doc in docs]}
# LLM 节点
def llm_node(state: AgentState):
"""调用大模型生成回复"""
prompt = create_prompt(state)
response = llm.invoke(prompt)
return {"messages": [response]}
# 工具节点处理
async def call_tools(state: AgentState):
"""根据消息决定是否调用工具"""
last_msg = state['messages'][-1]
if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
return {'tools_called': ['yes']}
return {}
4.3 组装工作流
from langgraph.graph import StateGraph, END
# 创建图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("retrieval", retrieval_node)
workflow.add_node("llm", llm_node)
workflow.add_node("tools", tool_node)
# 设置入口点
workflow.set_entry_point("llm")
# 添加边（带条件判断）
workflow.add_conditional_edges(
"llm",
should_call_tools, # 返回下一个节点的函数
{
"tools": "tools",
"retrieve": "retrieval",
"end": END
}
)
# 循环边：工具调用完回到 LLM
workflow.add_edge("tools", "llm")
# 编译
app = workflow.compile()
五、调试与可视化
5.1 保存图到文件
from langgraph.store.memory import InMemoryStore
# 配置持久化
store = InMemoryStore()
app = workflow.compile(checkpointer=store)
# 保存为 DOT 文件
with open("graph.dot", "w") as f:
f.write(app.get_graph().to_dot())
使用 Graphviz 查看可视化:
dot -Tpng graph.dot -o graph.png
5.2 打印执行日志
from IPython.display import Image, display
# 可视化图结构
image_data = app.get_graph(xray=True).draw_mermaid_png()
display(Image(image_data))
六、进阶技巧
6.1 实现记忆管理
# 限制上下文窗口大小
MAX_TOKENS = 4000
def manage_context(state: AgentState):
"""保持对话历史在合理范围内"""
total_tokens = sum(len(msg.content) for msg in state['messages'])
if total_tokens > MAX_TOKENS:
# 只保留最近的 N 条消息
state['messages'] = state['messages'][-20:]
return state
6.2 错误恢复机制
MAX_RETRIES = 3
def handle_errors(state: AgentState):
"""检测到错误时的处理逻辑"""
if state['error_count'] >= MAX_RETRIES:
return {"final_answer": "抱歉，我无法完成这个任务"}
return {"error_count": state['error_count'] + 1}
6.3 用户中断与确认
from langgraph.constants import Send
async def require_confirmation(state: AgentState):
"""在执行高危操作前请求用户确认"""
if requires_user_approval(state):
user_response = await get_user_input()
if user_response == "approve":
return [Send("execute_action", {})]
else:
return {"final_answer": "操作已取消"}
return []
七、生产环境注意事项
7.1 异步处理优化
import asyncio
async def process_with_timeout(state: AgentState, timeout=30):
try:
result = await asyncio.wait_for(
some_async_operation(state),
timeout=timeout
)
return result
except asyncio.TimeoutError:
return {"error": "操作超时"}
7.2 缓存机制
from functools import lru_cache
@lru_cache(maxsize=128)
def cached_retrieval(query_hash: str) -> List[str]:
"""缓存查询结果避免重复调用"""
return vector_db.search(query_hash)
7.3 监控与指标
import time
def instrumented_node(name: str):
"""带性能监控的节点包装器"""
async def wrapper(state: AgentState):
start = time.time()
result = await original_node(state)
duration = time.time() - start
logger.info(f"{name} 耗时:{duration:.2f}s")
return result
return wrapper
八、常见问题与解决方案
Q1: 无限循环怎么办？
A: 设置最大步数限制:
max_iterations = 10
app = workflow.compile(
recursion_limit=max_iterations
)
Q2: 状态不一致导致错误？
A: 使用类型检查:
from pydantic import BaseModel
class VerifiedState(BaseModel):
messages: List[HumanMessage]
retrieved_docs: List[str]
Q3: 如何处理外部 API 失败？
A: 实现重试逻辑:
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def call_external_api(data):
# 带重试的外部调用
pass
九、总结
LangGraph 为企业级 AI 应用提供了强大的图编排能力:
✅ 状态管理: 统一的 TypedDict 状态定义
✅ 流程控制: 灵活的条件边和循环
✅ 调试友好: 可视化的图和详细的日志
✅ 生产就绪: 支持持久化、监控、异常处理
开始你的第一个 LangGraph 应用吧！记住：先理解你的业务流程，再设计对应的图结构。
