LangGraph实战：构建本地多智能体代码生成与审查系统

LangGraph 是 LangChain 生态中用于构建复杂、有状态多智能体应用的核心框架。它不是一个独立的模型，而是一个编排工具，让你能用“图”的思维来设计和运行由多个 AI 智能体（Agent）和工具（Tool）协同工作的系统。简单说，它解决了单个 Agent 能力有限、任务流程复杂的问题，让你能像搭积木一样，组合多个 AI 智能体来完成写代码、查资料、审核、决策等一系列连贯动作。

这篇文章不讲虚的概念，直接带你上手。我们会拆解 LangGraph 的核心架构，用代码实战一个能自动分析需求、写代码、运行测试的多智能体系统。整个过程不依赖付费 API，完全基于开源模型（如 Ollama）在本地运行，重点在于理解其工作流（Workflow）设计、状态（State）管理和监督（Supervision）机制。无论你是想构建自动化开发助手、智能客服链还是数据分析流水线，这套方法都能直接复用。

1. 核心能力速览

2. 适用场景与使用边界

适合谁用？

• AI 应用开发者：希望超越简单的单轮问答，构建具有复杂逻辑链的 AI 应用。
• 技术团队：需要将 AI 能力嵌入到现有业务系统，实现自动化流程（如自动生成报告、代码审查、客户工单处理）。
• 研究者与学习者：希望深入理解多智能体系统（MAS）的工程化实现。

能解决什么问题？

1. 任务分解与协作：将一个复杂任务（如“开发一个网页爬虫”）自动分解为需求分析、架构设计、代码编写、单元测试等子任务，并由不同特长的智能体分工完成。
2. 有状态会话管理：在长时间的交互中（如客服对话、设计迭代），保持上下文连贯，记住之前所有的决策和输出。
3. 流程控制与纠错：当某个智能体输出不符合要求时，能自动路由到复核或重试节点，甚至引入人工审核。
4. 工具集成与调度：灵活调用搜索引擎、数据库、代码执行环境、外部 API 等多种工具。

不适合什么场景？

• 极其简单的单次问答：用 LangChain 的简单 Chain 或直接调用模型 API 更轻量。
• 对延迟极其敏感：多智能体间的多次调用和状态传递会增加整体响应时间。
• 完全静态、无分支的线性流程：可能过于复杂。

合规与安全边界：

• 责任归属：由多智能体自动生成的代码、文本等内容，开发者需承担最终审核责任，确保其合规、安全、无偏见。
• 数据隐私：工作流中传递的数据可能包含敏感信息，需注意加密存储和传输，避免泄露。
• 工具使用：智能体调用的工具（如代码执行、网络访问）具有潜在风险，必须在沙箱或严格权限控制下运行。

3. 环境准备与前置条件

我们将构建一个本地运行的多智能体系统，核心框架是 LangGraph，智能体大脑使用本地部署的 Ollama 模型。

基础环境清单：

• 操作系统：Windows 10/11, macOS, 或 Linux (推荐 Ubuntu 20.04+)
• Python：版本 3.10 或 3.11。避免使用 3.12 以防某些包兼容性问题。
• 包管理工具：pip或conda。
• 本地大模型服务：Ollama。用于提供本地 LLM 能力，我们将使用llama3.2或qwen2.5等轻量模型。
• 网络：能正常访问 PyPI 和 GitHub 以下载 Python 包。

可选但推荐的工具：

• 代码编辑器：VS Code 或 PyCharm。
• 虚拟环境：使用venv或conda创建独立环境，避免包冲突。
• LangGraph Studio：LangGraph 的可视化开发工具，方便调试。

4. 安装部署与启动方式

4.1 创建并激活虚拟环境

# 创建虚拟环境 python -m venv langgraph_env # 激活环境 (Windows) langgraph_env\Scripts\activate # 激活环境 (macOS/Linux) source langgraph_env/bin/activate

4.2 安装核心依赖

pip install langgraph langchain langchain-community ollama

• langgraph: 核心工作流编排框架。
• langchain: 提供智能体、链、提示词模板等基础组件。
• langchain-community: 包含大量第三方工具和模型集成。
• ollama: LangChain 对 Ollama 的官方集成包。

4.3 安装并启动 Ollama 服务

如果你还没有安装 Ollama，请先安装并拉取一个模型。

# 前往 https://ollama.com 下载并安装 Ollama # 拉取一个模型，例如 llama3.2 ollama pull llama3.2 # 启动 Ollama 服务，它默认会在 11434 端口提供 API # Ollama 会作为后台服务运行

4.4 验证 Ollama 服务

运行一个简单的 Python 脚本来测试 Ollama 连接是否正常。

# test_ollama.py from langchain_community.llms import Ollama llm = Ollama(model="llama3.2") response = llm.invoke("Hello, what is LangGraph?") print(response)

如果能看到模型返回的关于 LangGraph 的介绍文本，说明环境配置成功。

5. 功能测试与效果验证：构建一个代码生成与审查多智能体

我们将构建一个包含三个智能体的工作流：

1. 需求分析师 (Analyst)：理解用户模糊的需求，输出清晰、可执行的任务规格。
2. 程序员 (Coder)：根据任务规格，编写 Python 代码。
3. 审查员 (Reviewer)：审查代码，检查语法错误、逻辑问题，并提出修改建议。

5.1 定义共享状态 (State)

工作流中所有智能体都能读写这个状态。

# state.py from typing import TypedDict, List, Annotated import operator class State(TypedDict): # 用户原始输入 original_request: str # 需求分析师输出的清晰任务描述 clarified_spec: str # 程序员生成的代码 generated_code: str # 审查员提出的修改建议 review_comments: List[str] # 记录工作流执行到了哪一步 step: str

5.2 实现智能体节点 (Nodes)

每个智能体是一个函数，接收当前State，更新它，并返回更新后的State。

# nodes.py from langchain_core.prompts import ChatPromptTemplate from langchain_community.llms import Ollama from .state import State llm = Ollama(model="llama3.2") def analyst_node(state: State) -> State: """需求分析智能体""" prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个资深产品经理，擅长将模糊的需求转化为清晰、无歧义的技术任务描述。"), ("human", "用户的需求是：{request}\n\n请输出一份清晰的技术任务描述，包含输入、处理过程、输出示例。") ]) chain = prompt | llm clarified_spec = chain.invoke({"request": state["original_request"]}) # 更新状态 state["clarified_spec"] = clarified_spec state["step"] = "需求分析完成" print(f"[Analyst] 需求已澄清：{clarified_spec[:100]}...") return state def coder_node(state: State) -> State: """程序员智能体""" prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的 Python 程序员，严格按照给定的任务描述编写简洁、高效、可运行的代码。只输出代码块，不要解释。"), ("human", "任务描述：{spec}\n\n请编写完整的 Python 代码。") ]) chain = prompt | llm generated_code = chain.invoke({"spec": state["clarified_spec"]}) state["generated_code"] = generated_code state["step"] = "代码生成完成" print(f"[Coder] 代码已生成，长度：{len(generated_code)} 字符") return state def reviewer_node(state: State) -> State: """代码审查智能体""" prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个严格的代码审查员。检查代码的语法、逻辑、风格和潜在错误。如果代码优秀，说‘通过’；否则，列出具体的修改建议。"), ("human", "任务描述：{spec}\n\n待审查代码：\n```python\n{code}\n```") ]) chain = prompt | llm review_result = chain.invoke({"spec": state["clarified_spec"], "code": state["generated_code"]}) # 简单判断审查结果 comments = [review_result] if review_result.strip() != "通过" else [] state["review_comments"] = comments state["step"] = "代码审查完成" if comments: print(f"[Reviewer] 审查未通过，建议：{comments}") else: print("[Reviewer] 审查通过！") return state

5.3 构建工作流图 (Graph) 并设置路由逻辑

这是 LangGraph 的核心：定义节点如何连接。

# graph.py from langgraph.graph import StateGraph, END from .state import State from .nodes import analyst_node, coder_node, reviewer_node # 1. 创建图，并指定状态类型 workflow = StateGraph(State) # 2. 添加节点 workflow.add_node("analyst", analyst_node) workflow.add_node("coder", coder_node) workflow.add_node("reviewer", reviewer_node) # 3. 设置入口点 workflow.set_entry_point("analyst") # 4. 定义边（连接逻辑） workflow.add_edge("analyst", "coder") # 分析完直接去写代码 workflow.add_edge("coder", "reviewer") # 写完代码去审查 # 5. 定义条件边：根据审查结果决定是否结束 def decide_after_review(state: State) -> str: """审查后决策：如果有修改建议，则重新分析需求（模拟迭代）；否则结束。""" if state["review_comments"]: print("审查未通过，返回需求分析节点进行迭代...") return "analyst" # 返回分析师节点，开始新一轮迭代（实际中可能返回coder） else: return END # 结束工作流 workflow.add_conditional_edges( "reviewer", # 从哪个节点出发 decide_after_review, # 决策函数 { "analyst": "analyst", # 如果返回 "analyst"，则跳转到 analyst 节点 END: END # 如果返回 END，则结束 } ) # 6. 编译图，得到可执行对象 app = workflow.compile()

5.4 运行工作流进行测试

现在，我们可以用这个图来处理一个用户请求了。

# main.py from graph import app from state import State # 初始化状态 initial_state: State = { "original_request": "帮我写一个程序，能从网上抓取今日天气并保存到文件。", "clarified_spec": "", "generated_code": "", "review_comments": [], "step": "开始" } print("=== 开始执行多智能体工作流 ===") # 运行工作流 try: final_state = app.invoke(initial_state) print("\n=== 工作流执行完毕 ===") print(f"最终状态: {final_state['step']}") print(f"\n生成的代码:\n{final_state['generated_code']}") if final_state['review_comments']: print(f"\n审查意见（迭代后）: {final_state['review_comments']}") except Exception as e: print(f"工作流执行出错: {e}")

执行与观察：

1. 在终端运行python main.py。
2. 你会看到控制台依次打印[Analyst]、[Coder]、[Reviewer]的日志，清晰地展示了工作流的执行顺序。
3. 如果审查员第一次给出了修改建议 (review_comments非空)，根据我们设定的条件路由，工作流会跳回analyst节点，开始新一轮的“分析-编码-审查”循环，直到审查通过。
4. 最终，控制台会输出生成的 Python 代码。

效果验证点：

• 顺序执行：是否严格按照分析师 -> 程序员 -> 审查员的顺序执行？
• 状态传递：clarified_spec是否从分析师正确传递给了程序员和审查员？
• 条件路由：当review_comments不为空时，工作流是否真的回到了analyst节点？（可以在reviewer_node中故意返回一条评论来测试）。
• 输出质量：生成的代码是否基本符合“抓取天气并保存”的需求？虽然本地小模型能力有限，但流程是通的。

6. 接口 API 与批量任务

6.1 将工作流封装为 FastAPI 服务

将上述编译好的app暴露为 HTTP API，方便集成。

# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from graph import app from state import State app_fastapi = FastAPI(title="多智能体代码生成服务") class CodeRequest(BaseModel): original_request: str class CodeResponse(BaseModel): clarified_spec: str generated_code: str review_comments: list step: str @app_fastapi.post("/generate_code", response_model=CodeResponse) async def generate_code(request: CodeRequest): """接收用户需求，返回多智能体协作生成的代码""" initial_state: State = { "original_request": request.original_request, "clarified_spec": "", "generated_code": "", "review_comments": [], "step": "开始" } try: final_state = app.invoke(initial_state) return CodeResponse(**final_state) except Exception as e: raise HTTPException(status_code=500, detail=f"工作流执行失败: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app_fastapi, host="0.0.0.0", port=8000)

运行python api_server.py，服务将在http://127.0.0.1:8000启动。

6.2 调用 API 示例

使用curl或 Python 客户端进行测试。

# 使用 curl 调用 curl -X POST "http://127.0.0.1:8000/generate_code" \ -H "Content-Type: application/json" \ -d '{"original_request": "写一个函数，计算斐波那契数列的前N项。"}'

# 使用 Python requests 调用 import requests import json url = "http://127.0.0.1:8000/generate_code" payload = {"original_request": "写一个函数，计算斐波那契数列的前N项。"} headers = {"Content-Type": "application/json"} response = requests.post(url, data=json.dumps(payload), headers=headers) if response.status_code == 200: result = response.json() print(f"任务状态: {result['step']}") print(f"生成代码:\n{result['generated_code']}") else: print(f"请求失败: {response.status_code}, {response.text}")

6.3 批量任务处理

工作流本身是无状态的，状态由每次调用初始化。因此，处理批量任务非常简单：循环调用app.invoke()或并发调用 API。

# batch_processor.py from graph import app from state import State import concurrent.futures requests = [ "写一个Python脚本，重命名当前目录下所有.txt文件。", "写一个函数，验证电子邮件地址格式是否正确。", "写一个程序，从JSON文件中读取数据并生成柱状图。", ] def process_one_request(user_request: str): """处理单个请求""" initial_state: State = { "original_request": user_request, "clarified_spec": "", "generated_code": "", "review_comments": [], "step": "开始" } result = app.invoke(initial_state) return { "request": user_request, "code": result["generated_code"], "status": result["step"] } # 顺序处理 print("=== 顺序处理 ===") for req in requests: output = process_one_request(req) print(f"需求: {output['request'][:30]}... | 状态: {output['status']}") # 使用线程池并发处理（注意：如果后端是CPU/GPU密集型，需控制并发数） print("\n=== 并发处理 (4线程) ===") with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: future_to_req = {executor.submit(process_one_request, req): req for req in requests} for future in concurrent.futures.as_completed(future_to_req): req = future_to_req[future] try: output = future.result() print(f"需求: {output['request'][:30]}... | 状态: {output['status']}") except Exception as exc: print(f'{req} 生成时发生异常: {exc}')

关键点：批量任务时，主要瓶颈在于底层 LLM 的推理速度。需要根据 Ollama 服务器的能力（CPU/GPU）合理设置并发数，避免压垮服务。

7. 资源占用与性能观察

LangGraph 框架本身非常轻量，其资源消耗主要来自两方面：

1. Python 进程内存：用于维护工作流状态、消息传递等。通常很小（几十到几百MB）。
2. 大模型推理资源：这是主要开销。取决于你使用的 Ollama 模型大小和硬件。

性能观察方法：

1. Ollama 服务器监控：

   • 运行ollama serve的终端会显示每个请求的处理时间。
   • 可以使用ollama ps查看正在运行的模型实例。

2. 工作流执行时间：

   • 在代码中关键节点添加时间戳，计算每个智能体的耗时。

   import time def analyst_node(state: State) -> State: start = time.time() # ... 原有逻辑 ... end = time.time() print(f"分析师节点耗时: {end - start:.2f}秒") return state

3. 系统资源监控：

   • Linux/macOS: 使用htop或nvidia-smi（如有 GPU）查看 CPU/GPU 和内存占用。
   • Windows: 使用任务管理器。

影响性能的关键因素：

• 模型大小：llama3.2(3B参数) 比qwen2.5:14b快得多，但能力可能较弱。
• 提示词长度：输入给模型的提示词越长，推理耗时越长。
• 工作流复杂度：节点数量、条件分支数量、循环次数。每次循环都意味着重新调用模型。
• 硬件：GPU 推理远快于 CPU。

优化建议：

• 本地部署：使用量化后的模型（如llama3.2:3b-instruct-q4_K_M）平衡速度与质量。
• 缓存：对重复性高的子任务结果进行缓存。
• 剪枝：优化工作流，减少不必要的节点调用或循环。
• 异步调用：如果工作流中某些节点不依赖 LLM（如纯计算或数据库查询），可将其设计为异步。

8. 常见问题与排查方法

9. 最佳实践与使用建议

1. 从简单开始，逐步复杂化：先构建一个两三个节点的线性工作流并跑通，再逐步添加条件分支、循环和并行节点。
2. 精心设计状态结构：State是所有节点共享的“黑板”。设计时要想清楚哪些数据需要共享，类型是什么。避免状态过于庞大。
3. 编写清晰的提示词：多智能体的性能很大程度上取决于每个智能体的提示词。为每个角色（分析师、程序员、审查员）设计专业、明确的系统指令。
4. 实现健壮的错误处理：在工作流的关键节点（如调用外部 API、执行代码）添加try...except，并定义错误处理路径（例如，路由到一个“错误处理”节点）。
5. 引入人工干预点：对于关键决策或敏感操作，使用Human-in-the-loop机制，让工作流暂停并等待人工确认。LangGraph 对此有原生支持。
6. 使用 LangGraph Studio 进行调试：这是官方可视化工具，可以直观地看到工作流的执行路径、状态变化，极大提升开发效率。
7. 版本控制与测试：像管理普通代码一样管理你的工作流定义（graph.py、state.py），并为其编写单元测试，模拟不同的输入状态。
8. 性能与成本监控：在生产环境中，记录每个工作流实例的执行时间、调用模型的 Token 消耗（如果使用计费 API）和最终结果，用于分析和优化。
9. 安全与合规前置：如果工作流最终生成的内容（如代码、文本）将对外发布或使用，必须建立人工审核流程，确保其安全性、合规性和准确性。

10. 总结与下一步

LangGraph 为我们提供了一套强大而直观的“乐高积木”，用于搭建多智能体应用。它的核心优势在于将复杂流程可视化、模块化，并通过有状态图的概念统一管理上下文和流程控制。

通过本文的实战，你应该已经掌握了：

• 核心概念：状态（State）、节点（Node）、边（Edge）、条件路由（Conditional Edge）。
• 搭建流程：定义状态 → 实现节点函数 → 构建图并设置路由 → 编译运行。
• 本地部署：如何结合 Ollama 在本地运行完全开源的多智能体系统。
• 服务化：如何将工作流封装成 API，并处理批量任务。

最容易踩的坑：

• 忘记在节点函数中return state。
• 条件路由的逻辑错误导致死循环。
• 提示词不够精确，导致智能体行为偏离预期。

下一步可以探索的方向：

1. 集成更强大的模型：将 Ollama 节点替换为 OpenAI GPT、Claude 或国内大模型的 API 调用，以提升智能体能力。
2. 添加更多工具：让智能体能够调用搜索引擎、数据库、文件系统、代码解释器等，使其能力从“思考”扩展到“行动”。
3. 实现并行执行：使用langgraph.graph中的add_edge和add_conditional_edges组合，实现多个节点并行运行，提升效率。
4. 探索 LangGraph Studio：使用图形化界面拖拽构建工作流，这将是你开发复杂应用的神器。
5. 研究高级模式：如子图（Subgraph）、消息持久化、检查点（Checkpointing）等，用于构建更稳定、可恢复的长期运行应用。

多智能体系统是构建下一代 AI 应用的关键架构。利用 LangGraph 上手实践，是理解并掌握这一趋势的最佳途径。建议将本文的示例代码作为起点，不断修改和扩展，打造出适合你自己场景的自动化智能助手。