LangGraph实战：从零构建多智能体系统的图结构工作流
如果你正在为如何构建复杂的多智能体系统而头疼，觉得现有的 LangChain 框架在处理复杂工作流时力不从心，那么 LangGraph 的出现可能正是你需要的解决方案。与传统的线性链式调用不同，LangGraph 引入了图结构来编排智能体，让多角色协作、状态管理和循环控制变得前所未有的清晰。
这篇文章不会停留在概念介绍层面，而是通过完整的实战项目，带你从零构建一个真实可用的多智能体系统。你将学会如何用 LangGraph 设计智能体工作流、处理共享状态、实现条件分支，以及如何将理论应用到实际业务场景中。
1. 为什么 LangGraph 值得你投入时间学习？
在传统的 AI 应用开发中，我们往往面临一个困境：简单的任务可以用 LangChain 的链式调用解决，但一旦遇到需要多个智能体协作、有状态保持、或者需要根据中间结果动态调整流程的复杂场景，代码就会变得难以维护。
LangGraph 的核心价值在于它采用了图论的思想来建模智能体工作流。每个节点代表一个处理单元（智能体或工具），边代表控制流。这种设计带来了几个关键优势：
- 可视化的工作流：你可以清晰地看到整个系统的执行路径，便于调试和优化
- 灵活的状态管理：共享状态在节点间传递，避免了复杂的全局变量管理
- 支持循环和条件分支：这是传统链式结构难以实现的复杂逻辑
- 更好的错误处理：可以在特定节点设置重试机制和异常处理
从技术趋势来看，多智能体协作正在成为 AI 应用的主流范式。无论是企业级的客服系统、代码生成工具，还是复杂的决策支持系统，都需要多个专业化的智能体协同工作。LangGraph 为这种范式提供了工程上可行的实现方案。
2. LangGraph 核心概念解析
2.1 图（Graph）与节点（Node）
在 LangGraph 中，图是由节点和边组成的有向图。每个节点代表一个处理单元，可以是一个简单的函数，也可以是一个复杂的智能体。节点之间通过边连接，定义了数据流动的方向。
from langgraph.graph import Graph # 创建一个空的图 workflow = Graph() # 定义节点 def node_a(state): print("执行节点 A") return {"result": "A 完成"} def node_b(state): print("执行节点 B") return {"result": "B 完成"} # 添加节点到图中 workflow.add_node("node_a", node_a) workflow.add_node("node_b", node_b)2.2 状态（State）管理
LangGraph 的状态管理是其核心特性之一。状态是一个字典对象，在节点间传递和修改。你可以定义状态的 schema，确保类型安全。
from typing import TypedDict, Annotated from langgraph.graph import add_messages class State(TypedDict): messages: Annotated[list, add_messages] current_step: str intermediate_results: dict # 使用注解定义状态更新规则 def process_node(state: State) -> State: # 在这里处理状态 new_state = { "messages": state["messages"] + ["处理完成"], "current_step": "processed", "intermediate_results": {"data": "示例结果"} } return new_state2.3 边（Edge）与条件路由
边定义了节点之间的连接关系，可以是无条件跳转，也可以基于条件判断进行路由。
from langgraph.graph import END # 添加边定义节点执行顺序 workflow.add_edge("node_a", "node_b") # 条件路由示例 def should_continue(state: State) -> str: if state.get("need_more_processing", False): return "continue_processing" else: return END workflow.add_conditional_edges( "node_b", should_continue, { "continue_processing": "node_c", END: END } )3. 环境准备与依赖安装
3.1 Python 环境要求
LangGraph 需要 Python 3.8 或更高版本。建议使用虚拟环境来管理依赖：
# 创建虚拟环境 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # 或 langgraph-env\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain-openai3.2 模型配置
你需要配置 LLM 模型。这里以 OpenAI 为例，但 LangGraph 支持多种模型提供商：
from langchain_openai import ChatOpenAI import os # 设置 API 密钥 os.environ["OPENAI_API_KEY"] = "your-api-key-here" # 创建 LLM 实例 llm = ChatOpenAI(model="gpt-4")3.3 验证安装
创建一个简单的测试脚本来验证环境配置：
# test_environment.py from langgraph.graph import Graph from langchain_openai import ChatOpenAI def test_node(state): return {"status": "环境测试通过"} # 创建测试图 test_graph = Graph() test_graph.add_node("test", test_node) test_graph.set_entry_point("test") test_graph.set_finish_point("test") # 编译并测试 app = test_graph.compile() result = app.invoke({"status": "开始测试"}) print("测试结果:", result)运行测试脚本，确保一切正常。
4. 第一个 LangGraph 智能体实战
让我们从一个简单的多智能体系统开始：文档分析器。这个系统包含三个智能体：文档加载器、内容分析器和总结生成器。
4.1 定义状态 Schema
首先定义系统需要维护的状态：
from typing import TypedDict, Annotated, List from langgraph.graph import add_messages class DocumentState(TypedDict): document_content: str analysis_results: dict summary: str current_agent: str error_message: str4.2 实现各个智能体节点
def document_loader(state: DocumentState) -> DocumentState: """文档加载智能体""" print("文档加载器运行中...") # 模拟文档加载过程 sample_content = """ LangGraph 是一个用于构建多智能体系统的框架。 它基于图结构来编排智能体工作流，支持复杂的状态管理和条件路由。 主要特性包括：可视化工作流、灵活的状态管理、循环控制等。 """ return { **state, "document_content": sample_content, "current_agent": "document_loader" } def content_analyzer(state: DocumentState) -> DocumentState: """内容分析智能体""" print("内容分析器运行中...") content = state["document_content"] # 简单的分析逻辑 analysis = { "word_count": len(content.split()), "sentence_count": content.count('.') + content.count('!') + content.count('?'), "key_topics": ["多智能体系统", "图结构", "工作流编排"] } return { **state, "analysis_results": analysis, "current_agent": "content_analyzer" } def summary_generator(state: DocumentState) -> DocumentState: """总结生成智能体""" print("总结生成器运行中...") analysis = state["analysis_results"] summary = f""" 文档分析完成： - 总字数：{analysis['word_count']} - 句子数量：{analysis['sentence_count']} - 主要话题：{', '.join(analysis['key_topics'])} """ return { **state, "summary": summary, "current_agent": "summary_generator" }4.3 构建完整的工作流图
from langgraph.graph import Graph, END # 创建图实例 document_workflow = Graph() # 添加节点 document_workflow.add_node("loader", document_loader) document_workflow.add_node("analyzer", content_analyzer) document_workflow.add_node("summarizer", summary_generator) # 定义执行顺序 document_workflow.set_entry_point("loader") document_workflow.add_edge("loader", "analyzer") document_workflow.add_edge("analyzer", "summarizer") document_workflow.add_edge("summarizer", END) # 编译应用 app = document_workflow.compile()4.4 运行并测试
# 初始化状态 initial_state = { "document_content": "", "analysis_results": {}, "summary": "", "current_agent": "", "error_message": "" } # 执行工作流 result = app.invoke(initial_state) print("最终结果:", result["summary"])这个简单的例子展示了 LangGraph 的基本工作流程。接下来我们会深入更复杂的场景。
5. 高级特性：条件路由与循环控制
5.1 实现条件路由
在实际应用中，我们经常需要根据中间结果决定下一步执行路径。下面是一个智能客服系统的例子：
def intent_classifier(state: DocumentState) -> DocumentState: """意图分类智能体""" user_input = state.get("user_query", "") # 简单的意图分类逻辑 if "价格" in user_input or "多少钱" in user_input: intent = "price_inquiry" elif "技术支持" in user_input or "帮助" in user_input: intent = "technical_support" else: intent = "general_query" return {**state, "detected_intent": intent} def price_agent(state: DocumentState) -> DocumentState: """价格查询智能体""" return {**state, "response": "我们的产品价格是..."} def support_agent(state: DocumentState) -> DocumentState: """技术支持智能体""" return {**state, "response": "技术团队将尽快联系您..."} def general_agent(state: DocumentState) -> DocumentState: """通用查询智能体""" return {**state, "response": "请问您需要什么帮助？"} def route_based_on_intent(state: DocumentState) -> str: """基于意图的路由函数""" intent = state.get("detected_intent", "general_query") return intent # 构建条件路由图 support_workflow = Graph() support_workflow.add_node("classifier", intent_classifier) support_workflow.add_node("price", price_agent) support_workflow.add_node("support", support_agent) support_workflow.add_node("general", general_agent) support_workflow.set_entry_point("classifier") support_workflow.add_conditional_edges( "classifier", route_based_on_intent, { "price_inquiry": "price", "technical_support": "support", "general_query": "general" } ) support_workflow.add_edge("price", END) support_workflow.add_edge("support", END) support_workflow.add_edge("general", END)5.2 实现循环控制
有些场景需要循环执行直到满足特定条件，比如多轮对话：
def dialogue_manager(state: DocumentState) -> DocumentState: """对话管理智能体""" conversation_history = state.get("conversation", []) current_turn = len(conversation_history) + 1 # 模拟对话逻辑 if current_turn == 1: response = "您好！请问有什么可以帮您？" elif current_turn >= 5: # 限制对话轮数 response = "对话已达到最大轮数，感谢您的咨询！" else: response = f"这是第 {current_turn} 轮对话的回复" conversation_history.append({ "turn": current_turn, "response": response }) return {**state, "conversation": conversation_history} def should_continue_dialogue(state: DocumentState) -> str: """判断是否继续对话""" conversation = state.get("conversation", []) if len(conversation) >= 5: return "end" else: return "continue" # 构建循环对话图 dialogue_workflow = Graph() dialogue_workflow.add_node("manager", dialogue_manager) dialogue_workflow.set_entry_point("manager") dialogue_workflow.add_conditional_edges( "manager", should_continue_dialogue, { "continue": "manager", # 循环回到自身 "end": END } )6. 完整项目实战：智能代码审查系统
现在我们来构建一个完整的智能代码审查系统，包含多个专业智能体协作。
6.1 系统架构设计
系统包含以下智能体：
- 代码解析器：分析代码结构
- 语法检查器：检查语法错误
- 安全审查器：检测安全漏洞
- 性能分析器：评估代码性能
- 报告生成器：生成综合报告
6.2 实现各个智能体
class CodeReviewState(TypedDict): code_content: str language: str parsing_result: dict syntax_issues: list security_issues: list performance_metrics: dict final_report: str def code_parser(state: CodeReviewState) -> CodeReviewState: """代码解析智能体""" code = state["code_content"] language = state["language"] # 简单的解析逻辑 parsing_result = { "lines_of_code": len(code.split('\n')), "functions_count": code.count('def ') if language == "python" else code.count('function'), "imports_count": code.count('import ') if language == "python" else 0 } return {**state, "parsing_result": parsing_result} def syntax_checker(state: CodeReviewState) -> CodeReviewState: """语法检查智能体""" code = state["code_content"] language = state["language"] issues = [] # 简单的语法检查逻辑 if language == "python": if "print " in code and "(" not in code.split("print ")[1].split()[0]: issues.append("可能使用了 Python 2 风格的 print 语句") return {**state, "syntax_issues": issues} def security_auditor(state: CodeReviewState) -> CodeReviewState: """安全审查智能体""" code = state["code_content"] security_issues = [] # 简单的安全检测 dangerous_patterns = ["eval(", "exec(", "os.system("] for pattern in dangerous_patterns: if pattern in code: security_issues.append(f"检测到潜在危险模式: {pattern}") return {**state, "security_issues": security_issues} def performance_analyzer(state: CodeReviewState) -> CodeReviewState: """性能分析智能体""" parsing_result = state["parsing_result"] metrics = { "complexity_score": parsing_result["lines_of_code"] * 0.1, "maintainability_index": max(0, 100 - parsing_result["lines_of_code"] * 0.5) } return {**state, "performance_metrics": metrics} def report_generator(state: CodeReviewState) -> CodeReviewState: """报告生成智能体""" issues_count = len(state["syntax_issues"]) + len(state["security_issues"]) metrics = state["performance_metrics"] report = f""" 代码审查报告： - 代码行数：{state["parsing_result"]["lines_of_code"]} - 函数数量：{state["parsing_result"]["functions_count"]} - 发现问题：{issues_count} 个 - 复杂度评分：{metrics["complexity_score"]:.2f} - 可维护性指数：{metrics["maintainability_index"]:.2f} 详细问题： {chr(10).join(state["syntax_issues"] + state["security_issues"])} """ return {**state, "final_report": report}6.3 构建完整工作流
# 创建代码审查工作流 code_review_workflow = Graph() # 添加所有节点 code_review_workflow.add_node("parser", code_parser) code_review_workflow.add_node("syntax_checker", syntax_checker) code_review_workflow.add_node("security_auditor", security_auditor) code_review_workflow.add_node("performance_analyzer", performance_analyzer) code_review_workflow.add_node("report_generator", report_generator) # 设置执行顺序 code_review_workflow.set_entry_point("parser") code_review_workflow.add_edge("parser", "syntax_checker") code_review_workflow.add_edge("syntax_checker", "security_auditor") code_review_workflow.add_edge("security_auditor", "performance_analyzer") code_review_workflow.add_edge("performance_analyzer", "report_generator") code_review_workflow.add_edge("report_generator", END) # 编译应用 review_app = code_review_workflow.compile()6.4 测试代码审查系统
# 测试代码 test_code = """ import os def calculate_sum(a, b): result = a + b print result # Python 2 风格 return result def risky_function(): return eval("2 + 2") """ initial_state = { "code_content": test_code, "language": "python", "parsing_result": {}, "syntax_issues": [], "security_issues": [], "performance_metrics": {}, "final_report": "" } result = review_app.invoke(initial_state) print("审查结果:", result["final_report"])7. 性能优化与最佳实践
7.1 异步执行优化
对于 I/O 密集型的智能体，可以使用异步执行提高性能：
import asyncio async def async_document_loader(state: DocumentState) -> DocumentState: """异步文档加载""" await asyncio.sleep(0.1) # 模拟网络请求 return {**state, "content": "加载的文档内容"} async def async_processor(state: DocumentState) -> DocumentState: """异步处理器""" # 可以并行执行多个任务 tasks = [ async_document_loader(state), # 其他异步任务... ] results = await asyncio.gather(*tasks) return process_results(results)7.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 robust_agent(state: DocumentState) -> DocumentState: """带有重试机制的健壮智能体""" try: # 可能失败的操作 result = risky_operation(state) return {**state, "result": result} except Exception as e: print(f"操作失败: {e}") raise # 触发重试 def error_handler(state: DocumentState) -> DocumentState: """错误处理节点""" if "error" in state: return {**state, "response": "抱歉，处理过程中出现了问题"} return state7.3 状态序列化与持久化
对于长时间运行的工作流，需要状态持久化：
import json import pickle def save_state(state: DocumentState, filepath: str): """保存状态到文件""" with open(filepath, 'w') as f: json.dump(state, f, indent=2) def load_state(filepath: str) -> DocumentState: """从文件加载状态""" with open(filepath, 'r') as f: return json.load(f) # 在关键节点添加状态保存 def checkpoint_agent(state: DocumentState) -> DocumentState: """检查点智能体""" save_state(state, "checkpoint.json") return state8. 常见问题与解决方案
8.1 状态管理问题
问题：状态在不同节点间传递时出现数据丢失或类型错误
解决方案：
- 使用 TypedDict 明确状态结构
- 在每个节点返回时使用
{**state, "new_field": value}模式 - 添加状态验证逻辑
def validate_state(state: DocumentState) -> bool: """状态验证""" required_fields = ["document_content", "current_agent"] return all(field in state for field in required_fields)8.2 循环控制问题
问题：工作流陷入无限循环
解决方案：
- 设置最大迭代次数
- 添加超时机制
- 实现循环退出条件检查
def safe_loop_condition(state: DocumentState) -> str: """安全的循环条件检查""" iteration_count = state.get("iteration", 0) if iteration_count >= 10: # 最大迭代次数 return "end" # 业务逻辑判断 if state.get("completion_status") == "done": return "end" return "continue"8.3 性能瓶颈问题
问题：复杂工作流执行缓慢
解决方案：
- 使用异步执行
- 并行化独立节点
- 缓存昂贵操作的结果
- 优化智能体的实现逻辑
9. 生产环境部署建议
9.1 监控与日志
在生产环境中，完善的监控是必不可少的：
import logging from datetime import datetime # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger("langgraph_app") def logged_agent(state: DocumentState) -> DocumentState: """带有日志记录的智能体""" start_time = datetime.now() agent_name = state.get("current_agent", "unknown") logger.info(f"开始执行智能体: {agent_name}") try: # 业务逻辑 result = process_business_logic(state) duration = (datetime.now() - start_time).total_seconds() logger.info(f"智能体 {agent_name} 执行完成，耗时: {duration:.2f}s") return result except Exception as e: logger.error(f"智能体 {agent_name} 执行失败: {e}") raise9.2 配置管理
使用环境变量或配置文件管理敏感信息：
import os from dataclasses import dataclass @dataclass class AppConfig: openai_api_key: str max_iterations: int log_level: str def load_config() -> AppConfig: """加载配置""" return AppConfig( openai_api_key=os.getenv("OPENAI_API_KEY"), max_iterations=int(os.getenv("MAX_ITERATIONS", "10")), log_level=os.getenv("LOG_LEVEL", "INFO") )9.3 版本控制与回滚
对于重要的工作流，实现版本管理：
import hashlib def get_workflow_version(workflow_config: dict) -> str: """计算工作流配置的版本哈希""" config_str = json.dumps(workflow_config, sort_keys=True) return hashlib.md5(config_str.encode()).hexdigest()[:8] # 保存版本信息 workflow_version = get_workflow_version(workflow_config) logger.info(f"工作流版本: {workflow_version}")通过本文的实战教程，你应该已经掌握了 LangGraph 的核心概念和实际应用技巧。从简单的线性工作流到复杂的多智能体系统，LangGraph 提供了一个强大而灵活的框架来构建下一代 AI 应用。
关键是要理解图思维：将复杂问题分解为相互协作的节点，通过清晰的数据流和控制流来组织整个系统。这种思维方式不仅适用于 LangGraph，对于任何复杂的系统设计都有借鉴意义。
建议从一个小项目开始实践，逐步增加复杂度。在实际项目中，你会遇到更多具体问题，但有了这个基础，你就能更好地理解和解决它们。