企业官网建设流程全解析

这次我们来看一个关于 LangGraph 智能体开发的实战教程。LangGraph 作为 LangChain 生态中用于构建复杂、有状态多步骤应用（智能体）的框架，正成为连接大模型 API 与具体业务逻辑的关键桥梁。它的核心价值在于，让你能用清晰的图（Graph）结构来定义智能体的决策流和工作流，告别传统 prompt 工程中状态管理混乱的痛点。

对于开发者而言，最关心的问题通常是：它能不能快速集成现有的大模型 API（如 OpenAI、DeepSeek、智谱等）？代码结构是否清晰易上手？能否处理需要记忆和工具调用的复杂任务？以及，它和 LangChain 到底是什么关系，该用哪个？这篇文章将直接切入这些核心问题，通过架构剖析和代码实战，带你快速掌握 LangGraph 构建智能体的核心方法。无论你是想开发一个自动化的数据分析助手，还是一个能调用外部 API 的客服机器人，这里的内容都能提供直接的参考。

本文将围绕以下几个核心部分展开：首先，快速梳理 LangGraph 的核心概念与适用场景；然后，深入其架构中的关键组件（State、Node、Edge）；接着，通过一个完整的代码示例，演示如何构建一个具备工具调用和记忆能力的智能体；最后，会讨论如何将其部署为 API 服务，并分享开发中的常见问题与调试技巧。目标是让你看完后，能立即动手搭建自己的第一个 LangGraph 智能体。

1. 核心能力速览

在深入细节之前，我们先通过一个表格快速了解 LangGraph 的核心特性和能力边界，这有助于判断它是否适合解决你手头的问题。

2. 适用场景与使用边界

LangGraph 并非所有 AI 应用的首选。理解其适用边界，能帮你更高效地选择技术栈。

它非常适合以下场景：

• 复杂多轮对话系统：客服机器人需要根据历史对话、用户意图和知识库检索结果，决定是回答问题、转人工还是询问更多信息。
• 自动化工作流：一个数据分析智能体，需要依次执行“接收查询 -> 规划分析步骤 -> 查询数据库 -> 执行计算 -> 生成图表 -> 撰写报告”。
• 游戏与模拟：NPC 需要根据环境状态、玩家交互和自身目标，规划一系列动作（移动、对话、使用物品）。
• 研究/写作助手：智能体接收一个主题，然后执行“搜索资料 -> 总结要点 -> 起草大纲 -> 撰写章节 -> 润色修改”的流程。

它可能不是最佳选择，或需要搭配其他技术的场景：

• 简单的单次问答（Q&A）：如果只是输入问题，模型直接输出答案，使用 LangChain 的LLMChain或直接调用模型 API 更简单。
• 单纯的文本生成/翻译：没有复杂状态流转和工具调用，用基础模型接口即可。
• 对延迟极其敏感的实时应用：图结构的调度会引入额外开销，纯推理场景应追求最小化延迟。
• 完全无状态的批处理任务：如果任务只是对一批独立数据做相同处理，用常规脚本或并行计算框架更直接。

合规与安全边界提醒：当使用 LangGraph 构建智能体时，尤其是涉及以下能力，必须特别注意：

1. 工具调用：智能体可以调用外部 API、操作数据库、执行代码。必须严格限制其权限，避免未授权访问或破坏性操作。在生产环境中，应对工具进行沙箱化处理或严格的输入校验。
2. 长期记忆：如果智能体存储用户对话或数据，需明确隐私政策，遵守数据保护法规（如 GDPR），并提供数据清除机制。
3. 内容生成：确保智能体生成的内容符合法律法规和平台政策，避免产生侵权、虚假或有害信息。建议增加内容过滤层。
4. 模型选择：所选大模型 API 的服务条款和内容政策同样适用于你的智能体。

3. 环境准备与前置条件

开始 LangGraph 实战之前，需要准备好开发环境。以下是一个通用的环境清单，你可以根据实际项目进行调整。

基础环境要求：

• 操作系统：Windows 10/11, macOS, 或 Linux (推荐 Ubuntu 20.04+)。
• Python：版本 3.8 或更高。建议使用 3.9+ 以获得最佳兼容性。
• 包管理工具：pip或conda。

核心依赖安装：LangGraph 通常与 LangChain 一起使用。我们将安装核心的 LangChain 包、LangGraph 以及一个用于与大模型交互的社区包。这里以使用 OpenAI API 为例。

# 创建并激活一个虚拟环境（强烈推荐） python -m venv langgraph-env # Windows: langgraph-env\Scripts\activate # macOS/Linux: source langgraph-env/bin/activate # 升级 pip pip install --upgrade pip # 安装核心依赖 pip install langgraph langchain langchain-openai # 可选：安装用于构建API服务的框架 pip install fastapi uvicorn

API Key 配置：你需要准备大模型服务的 API Key。这里以 OpenAI 为例，你需要一个有效的 OpenAI API Key。

• 将 Key 设置为环境变量是最安全、通用的做法。

  # Windows (PowerShell) $env:OPENAI_API_KEY="your-api-key-here" # macOS/Linux export OPENAI_API_KEY="your-api-key-here"

• 也可以在代码中直接设置，但不推荐将密钥硬编码在源码中。

验证安装：创建一个简单的 Python 脚本test_env.py来测试环境是否就绪。

import os from langchain_openai import ChatOpenAI # 确保已设置环境变量 OPENAI_API_KEY llm = ChatOpenAI(model="gpt-3.5-turbo") try: response = llm.invoke("Hello, world!") print("环境验证成功！") print(f"模型回复: {response.content}") except Exception as e: print(f"环境验证失败，错误信息: {e}") print("请检查：1. API Key 2. 网络连接 3. 依赖包是否安装正确")

运行此脚本，如果看到成功的回复，说明基础环境已配置完成。

4. LangGraph 核心架构与概念解析

理解 LangGraph 的架构是高效使用它的关键。其核心思想是将智能体的执行流程建模为一个有向图。下面我们拆解其中的核心组件。

4.1 State（状态）

State 是贯穿整个图执行过程的“共享内存”。它定义了智能体工作流中需要传递和更新的所有数据。State 通常是一个 TypedDict 或 Pydantic 模型。

from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): # 用户输入的问题 input: str # 智能体生成的中间思考或最终答案 output: str # 对话历史记录 chat_history: list # 工具调用的结果列表 tool_results: Annotated[list, operator.add] # 这是一个“归约器”，用于追加列表

Annotated用于声明某个字段的“归约”方式。operator.add表示当多个节点修改tool_results时，其结果会被追加（append）到列表中，而不是覆盖。这是实现“记忆”或“收集结果”的关键机制。

4.2 Node（节点）

Node 是图中的一个执行单元，是一个普通的 Python 函数（或可调用对象）。它接收当前的 State，执行一些操作（如调用 LLM、使用工具），然后返回一个包含 State 更新内容的字典。

def call_model(state: AgentState): """节点：调用大模型生成思考或回答""" from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo") # 构建提示词，可以包含历史、问题等来自 State 的信息 messages = [ ("system", "你是一个有帮助的助手。"), ("human", state["input"]) ] # 调用模型 response = llm.invoke(messages) # 返回要更新到 State 的内容 return {"output": response.content} def use_tool(state: AgentState): """节点：调用某个工具（例如查询天气）""" # 假设我们有一个工具函数 get_weather tool_result = get_weather(state["input"]) # 返回结果，tool_results 字段会因为归约器而追加这个结果 return {"tool_results": [tool_result]}

4.3 Edge（边）

Edge 定义了节点之间的流转逻辑。它决定了一个节点执行完毕后，下一个该执行哪个节点。边可以是固定的，也可以是基于 State 内容的条件边。

• 固定边：graph.add_edge(“node_a”, “node_b”)表示node_a执行完后总是执行node_b。
• 条件边：通过graph.add_conditional_edges添加，它需要一个路由函数，根据 State 返回下一个节点的名称。

def should_use_tool(state: AgentState) -> str: """路由函数：判断是否需要调用工具""" output = state.get("output", "") # 简单逻辑：如果模型输出中包含“查询天气”关键词，则路由到工具节点 if "天气" in output: return "use_tool_node" else: return "end_node" # 否则结束

4.4 Graph（图）与 Compilation（编译）

将节点和边组装起来，就形成了一个StateGraph。最后，需要调用graph.compile()来生成一个可执行的、优化过的CompiledGraph对象，这才是我们实际运行的东西。

from langgraph.graph import StateGraph, END # 1. 创建图，并指定 State 的类型 workflow = StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“call_model”, call_model) workflow.add_node(“use_tool”, use_tool) # 3. 设置入口点 workflow.set_entry_point(“call_model”) # 4. 添加边和条件边 workflow.add_conditional_edges( “call_model”, should_use_tool, # 路由函数 { “use_tool_node”: “use_tool”, # 如果返回 “use_tool_node”，则跳转到 use_tool 节点 “end_node”: END # 如果返回 “end_node”，则结束图执行 } ) workflow.add_edge(“use_tool”, END) # 工具节点执行完后结束 # 5. 编译图 app = workflow.compile()

编译后的app就是一个可以调用的智能体。你可以通过app.invoke(initial_state)来运行它。

5. 实战：构建一个具备工具调用能力的智能体

现在，我们将构建一个完整的智能体。它的功能是：回答用户问题，当问题涉及“计算”时，调用一个计算器工具；否则，直接由模型回答。

5.1 定义 State 和工具

from typing import TypedDict, Annotated, List import operator from langchain_core.tools import tool # 1. 定义 State class CalculatorState(TypedDict): question: str # 用户问题 model_thought: str # 模型的思考过程 final_answer: str # 给用户的最终答案 used_tools: Annotated[List[str], operator.add] # 记录使用了哪些工具 # 2. 定义一个简单的计算器工具 @tool def calculator(expression: str) -> str: """计算一个数学表达式的值。支持加减乘除和括号。例如：`(3 + 5) * 2`""" try: # 警告：实际生产中应对表达式进行严格安全检查，避免代码注入 result = eval(expression) return f“计算器结果：{expression} = {result}” except Exception as e: return f“计算错误：{e}” # 工具列表 tools = [calculator]

5.2 创建节点函数

我们需要两个关键节点：一个负责让模型决定行动（思考），另一个负责执行工具。

from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent from langchain.agents.format_scratchpad import format_log_to_str from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder # 初始化模型 llm = ChatOpenAI(model=“gpt-3.5-turbo”, temperature=0) # 构建 ReAct 风格的提示词模板 prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个聪明的助手，可以回答问题或使用计算器工具。请严格按照以下格式回复：\nThought: 你的思考过程\nAction: 要使用的工具名，如果是最终答案则填 ‘Final Answer’\nAction Input: 工具的输入参数，如果是最终答案则直接给出答案”), (“user”, “{input}”), MessagesPlaceholder(variable_name=“agent_scratchpad”), ]) # 节点1：代理（Agent）节点 - 决定思考、行动或直接回答 def agent_node(state: CalculatorState): # 将工具绑定到LLM llm_with_tools = llm.bind_tools(tools) # 创建ReAct智能体 agent = create_react_agent(llm_with_tools, tools, prompt) agent_executor = agent | ReActSingleInputOutputParser() # 调用智能体，传入当前问题 result = agent_executor.invoke({“input”: state[“question”]}) # 解析结果 if hasattr(result, ‘tool’) and result.tool: # 如果需要调用工具 return { “model_thought”: result.log, “action”: result.tool, “action_input”: result.tool_input } else: # 如果是最终答案 return { “model_thought”: result.log, “final_answer”: result.return_values[‘output’] } # 节点2：工具执行节点 def tool_node(state: CalculatorState): # 从state中获取要执行的动作 tool_name = state[“action”] tool_input = state[“action_input”] # 查找对应的工具 tool_map = {tool.name: tool for tool in tools} tool_to_use = tool_map[tool_name] # 执行工具 observation = tool_to_use.invoke(tool_input) # 更新状态，记录工具使用和观察结果 return { “used_tools”: [tool_name], “tool_observation”: observation }

5.3 构建并运行图

from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint import MemorySaver # 创建图 workflow = StateGraph(CalculatorState) # 添加节点 workflow.add_node(“agent”, agent_node) workflow.add_node(“execute_tool”, tool_node) # 设置入口点 workflow.set_entry_point(“agent”) # 定义边逻辑：根据 agent 节点的输出决定下一步 def route_after_agent(state: CalculatorState): # 如果 agent 节点产生了 final_answer，则结束 if “final_answer” in state and state[“final_answer”]: return “end” # 否则，说明需要执行工具 else: return “continue_to_tool” workflow.add_conditional_edges( “agent”, route_after_agent, { “continue_to_tool”: “execute_tool”, “end”: END } ) # 工具执行完后，应回到 agent 节点进行下一步思考（ReAct循环） workflow.add_edge(“execute_tool”, “agent”) # （可选）添加记忆检查点，使智能体能在多轮对话中记住历史 memory = MemorySaver() app = workflow.compile(checkpointer=memory) # 运行智能体！ initial_state = {“question”: “请问 (12 + 8) * 3 等于多少？”} final_state = app.invoke(initial_state, config={“configurable”: {“thread_id”: “test-1”}}) print(“用户问题：”, final_state[“question”]) print(“模型思考：”, final_state.get(“model_thought”)) print(“使用的工具：”, final_state.get(“used_tools”, [])) print(“最终答案：”, final_state.get(“final_answer”))

运行上述代码，你会看到智能体经历了 “Thought -> Action -> Tool -> Observation -> Thought -> Final Answer” 的完整 ReAct 循环，最终给出了正确答案。这个图结构清晰地定义了智能体的决策流。

6. 部署为 API 服务与批量任务处理

将 LangGraph 智能体封装成 API 服务，是将其集成到 Web 应用或其他系统的标准做法。我们使用 FastAPI 来实现。

6.1 创建 FastAPI 应用

# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import asyncio from your_langgraph_module import app # 导入之前编译好的 LangGraph app # 定义请求和响应模型 class AgentRequest(BaseModel): question: str thread_id: Optional[str] = None # 用于支持多轮对话的会话ID class AgentResponse(BaseModel): thread_id: str final_answer: str used_tools: list[str] model_thought: Optional[str] = None # 创建 FastAPI 实例 fastapi_app = FastAPI(title=“LangGraph 智能体 API”) @fastapi_app.post(“/chat”, response_model=AgentResponse) async def chat_with_agent(request: AgentRequest): try: # 准备初始状态 initial_state = {“question”: request.question} config = {“configurable”: {“thread_id”: request.thread_id or “default-thread”}} # 调用编译好的图。注意：invoke 是同步的，在异步环境中使用 run_in_executor 避免阻塞 loop = asyncio.get_event_loop() final_state = await loop.run_in_executor( None, app.invoke, initial_state, config ) # 构建响应 response = AgentResponse( thread_id=config[“configurable”][“thread_id”], final_answer=final_state.get(“final_answer”, “未生成答案”), used_tools=final_state.get(“used_tools”, []), model_thought=final_state.get(“model_thought”) ) return response except Exception as e: raise HTTPException(status_code=500, detail=f“智能体执行失败: {str(e)}”) @fastapi_app.get(“/health”) async def health_check(): return {“status”: “healthy”}

6.2 启动服务

使用 Uvicorn 启动这个 FastAPI 应用。

uvicorn main:fastapi_app --host 0.0.0.0 --port 8000 --reload

启动后，你可以通过http://localhost:8000/docs访问自动生成的 API 文档，并测试/chat接口。

6.3 批量任务处理

对于批量处理大量问题，简单的循环调用 API 可能效率低下且难以管理。我们可以设计一个简单的异步批量处理器。

# batch_processor.py import aiohttp import asyncio from typing import List import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) async def process_single_question(session: aiohttp.ClientSession, api_url: str, question: str, thread_id: str): """处理单个问题""" payload = {“question”: question, “thread_id”: thread_id} try: async with session.post(api_url, json=payload) as response: if response.status == 200: result = await response.json() logger.info(f“成功处理: ‘{question}’ -> {result[‘final_answer’][:50]}...”) return result else: error_text = await response.text() logger.error(f“处理失败 ‘{question}’: HTTP {response.status} - {error_text}”) return None except Exception as e: logger.error(f“请求异常 ‘{question}’: {e}”) return None async def batch_process_questions(questions: List[str], api_url: str = “http://localhost:8000/chat”, max_concurrent: int = 5): """批量处理问题列表，控制并发数""" connector = aiohttp.TCPConnector(limit=max_concurrent) timeout = aiohttp.ClientTimeout(total=60) async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session: tasks = [] for idx, q in enumerate(questions): # 为每个问题生成一个唯一的 thread_id，或根据业务逻辑决定 thread_id = f“batch-{idx}” task = asyncio.create_task(process_single_question(session, api_url, q, thread_id)) tasks.append(task) # 等待所有任务完成 results = await asyncio.gather(*tasks, return_exceptions=True) # 整理结果 successful_results = [] for q, r in zip(questions, results): if isinstance(r, Exception): logger.error(f“任务异常 for ‘{q}’: {r}”) elif r is not None: successful_results.append(r) logger.info(f“批量处理完成。总计: {len(questions)}, 成功: {len(successful_results)}”) return successful_results # 使用示例 if __name__ == “__main__”: question_list = [ “计算 25 * 4 的值”， “北京今天的天气怎么样？”， # 这个可能需要其他工具，此处仅为示例 “(18 - 7) / 2 等于多少？” ] # 运行批量处理 final_results = asyncio.run(batch_process_questions(question_list, max_concurrent=3)) for res in final_results: print(res)

这个批量处理器使用了aiohttp进行异步 HTTP 调用，并通过max_concurrent参数控制并发数，避免对 API 服务造成过大压力。

7. 资源占用、性能观察与优化

LangGraph 本身是一个轻量级的编排框架，其资源占用主要取决于：

1. 大模型 API 调用：这是主要的耗时和成本来源。响应时间取决于模型提供商和网络。
2. 工具执行：如果你集成了计算密集型或网络 I/O 密集型的工具，它们会成为瓶颈。
3. Python 运行时：图本身和状态管理开销通常很小。

性能观察点：

• 延迟：使用time模块记录app.invoke()的总耗时，拆分为模型调用耗时和工具执行耗时。
• API 调用次数：监控大模型 API 的调用次数和 Token 消耗，这是成本核心。
• 状态大小：如果 State 中存储了大量历史消息或中间结果，可能会影响内存和序列化/反序列化性能。

优化建议：

1. 精简 State：只保留必要的数据在 State 中，避免存储过大的对象（如图片二进制数据）。
2. 异步工具：如果工具涉及网络请求（如查询数据库、调用外部 API），将其实现为异步函数，并在图中使用异步节点，可以显著提升吞吐量。
3. 缓存：对于重复性的工具调用或模型查询结果，可以考虑引入缓存（如functools.lru_cache或 Redis）。
4. 流式输出：如果最终答案是文本，且模型支持，可以使用流式响应（Streaming）来提升用户体验。
5. 检查点（Checkpoint）优化：如果使用了MemorySaver等检查点，注意其存储后端。对于高并发，可能需要使用数据库（如 PostgreSQL）替代内存存储。

8. 常见问题与排查方法

在开发 LangGraph 智能体时，你可能会遇到以下典型问题。

9. 最佳实践与进阶建议

掌握了基础之后，遵循一些最佳实践能让你的 LangGraph 项目更加健壮和可维护。

1. State 设计要精简而明确：State 是你的智能体的“内存”。只定义工作流真正需要共享和传递的数据。使用Annotated和归约器（如operator.add）来优雅地处理列表追加等操作。
2. 节点函数保持纯净与可测试：每个节点函数应尽可能只做一件事。避免在节点内部处理复杂的副作用。这样便于单元测试和调试。
3. 充分利用可视化调试：LangGraph 提供了app.get_graph().draw_mermaid()功能，可以生成 Mermaid 图表，直观展示你的工作流。这在调试复杂图时非常有用。
4. 为生产环境配置检查点：开发时可以用MemorySaver，但生产环境建议使用持久化存储，如SqliteSaver或PostgresSaver，以确保会话状态不会丢失。
5. 实现严格的错误处理与回退：在节点中，对模型调用和工具调用进行try...catch。可以设计一个专门的“错误处理”节点，当任何节点失败时，路由到这里进行清理或提供友好回复。
6. 安全第一：尤其是工具调用。永远不要盲目执行模型生成的代码或系统命令。对工具输入进行严格的验证、清洗和权限控制。考虑在沙箱环境中运行不可信的工具。
7. 版本化你的图：当智能体逻辑更新时，图的定义可能改变。考虑如何管理不同版本的图，以及如何迁移已保存的检查点状态。
8. 监控与日志：在关键节点添加详细的日志记录，记录 State 的变化、工具调用结果和模型响应。这有助于追踪问题和分析智能体的决策过程。

LangGraph 将智能体的开发从“写一大段脆弱的 Prompt”变成了“组装可维护、可测试的组件”。它可能不是最简单上手的，但对于需要清晰逻辑和状态管理的复杂 AI 应用来说，它提供的结构和灵活性是无可替代的。从本文的简单计算器智能体出发，你可以尝试集成更复杂的工具（网络搜索、数据库、代码解释器）、设计多智能体协作的图、或者构建具备长期记忆和个性化能力的对话系统。