企业官网建设流程全解析

最近在尝试将大语言模型（LLM）从简单的问答工具升级为能自主执行复杂任务的智能体时，你是否也遇到了这些困扰：Agent 的逻辑流难以编排，状态管理混乱，记忆能力薄弱导致对话上下文丢失？传统的 LangChain 在构建复杂工作流时，代码往往变得冗长且难以维护。

本文将为你系统性地拆解 LangGraph 这一新兴的 Agent 编排框架，并结合 LangChain 生态，手把手带你从零构建具备长期记忆和复杂推理能力的 AI 智能体。无论你是希望入门 AI 应用开发的新手，还是寻求项目落地的进阶开发者，都能从中获得一套完整、可复现的实战方案。

1. 背景与核心概念：为什么需要 LangGraph？

在深入代码之前，我们首先要厘清几个核心概念以及它们之间的关系，这有助于理解 LangGraph 要解决的根本问题。

AI Agent（智能体）是什么？简单来说，它是一个能感知环境、进行决策并执行行动以实现目标的软件实体。在大模型语境下，Agent 通常由一个大语言模型（LLM）作为“大脑”，配合工具调用（Tools）、记忆（Memory）和任务规划（Planning）等模块构成。它不再是被动地回答单次提问，而是能够主动使用搜索引擎、计算器、数据库等工具，完成一系列连贯的任务，比如“帮我分析上周的销售数据并写一份报告”。

LangChain是一个用于开发由 LLM 驱动的应用程序的框架。它提供了丰富的模块，如模型抽象、提示模板、链（Chains）、记忆存储和工具集成，极大地简化了与 LLM 交互的复杂度。其核心抽象“链”可以将多个步骤串联起来。

然而，当任务流程不再是简单的线性链，而是包含条件分支、循环、并行执行等复杂逻辑时，传统的链式结构就会显得力不从心。代码会充斥着大量的if-else语句和状态管理，可读性和可维护性急剧下降。

这正是LangGraph登场的原因。LangGraph 是建立在 LangChain 之上的一个库，它引入了图（Graph）的概念来编排 Agent 的工作流。你可以将工作流中的每个步骤（节点）和步骤之间的流转逻辑（边）可视化地定义出来，从而清晰、灵活地构建出支持循环、分支和并行化的复杂 Agent。

核心概念对比与关系：

• LangChain： 提供构建 LLM 应用的基础“砖块”（模型、提示、工具、记忆）。
• LangGraph： 提供将这些“砖块”组装成复杂“机器”（智能体）的“蓝图”和“装配线”（图结构）。
• Agent： 最终构建出的、能够自主工作的智能应用程序。
• Memory： Agent 的“记忆”系统，用于在多次调用或不同节点间持久化状态（如对话历史、中间结果），是构建连贯性智能体的关键。

简单理解：LangChain + LangGraph = 强大的、可编排的 AI Agent。

2. 环境准备与版本说明

在开始实战前，我们需要搭建好开发环境。本文将使用 Python 作为开发语言，并优先考虑本地化部署方案以方便调试和隐私保护。

2.1 基础环境配置

请确保你的系统已安装 Python（推荐 3.8 及以上版本）。我们将使用venv创建独立的虚拟环境，避免包依赖冲突。

# 1. 创建项目目录并进入 mkdir langgraph-agent-tutorial && cd langgraph-agent-tutorial # 2. 创建虚拟环境（Windows 用户使用 `python -m venv venv`） python3 -m venv venv # 3. 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate # 4. 激活后，命令行提示符前应显示 (venv)

2.2 依赖安装

我们将安装 LangChain、LangGraph 的核心库，并选用Ollama作为本地大模型运行工具，它可以让您在本地无GPU或轻量GPU环境下运行如 Llama 3、Qwen 等开源模型。

# 安装核心框架 pip install langchain langgraph langchain-community # 安装 Ollama 的 LangChain 集成包（用于调用本地模型） pip install langchain-ollama # 安装用于结构化输出的可选包，对 Agent 很有用 pip install langchain-openai # 即使不用 OpenAI，它也包含一些有用的工具

版本说明（截至撰写时）：

• langchain: >=0.1.0 (LangChain 已进入 0.x 时代，API 与旧版有较大变化)
• langgraph: >=0.0.20
• langchain-ollama: >=0.1.0
• ollama软件：需单独在 Ollama官网 下载安装并拉取模型。

重要提示：LangChain 生态版本迭代较快，本文代码基于上述较新版本编写。若遇到 API 不兼容，请参考官方文档或适当调整版本号（例如pip install langchain==0.1.0）。

2.3 本地模型准备（Ollama）

1. 前往 Ollama官网 下载并安装对应操作系统的软件。
2. 打开终端，拉取一个合适的模型。这里我们选择轻量且性能不错的llama3.2:1b（10亿参数）作为示例，对硬件要求极低。

   ollama pull llama3.2:1b

3. 运行模型服务。Ollama 默认会在11434端口启动一个本地 API 服务。

   ollama run llama3.2:1b # 在另一个终端窗口保持服务运行，或直接让它在后台运行。

至此，你的开发环境已经就绪。

3. LangGraph 核心原理解析

LangGraph 的核心思想是将工作流抽象为一个有状态图（Stateful Graph）。理解以下几个关键组件是构建智能体的基础：

3.1 状态（State）

状态是一个字典（或 Pydantic 模型），它随着工作流的执行而演变，包含了所有节点需要共享和修改的信息。例如，它可能包含：

• messages: 对话消息列表（来自用户和AI）。
• intermediate_steps: 工具调用及其结果的记录。
• next: 指示下一步该执行哪个节点。
• 任何你自定义的业务数据。

3.2 节点（Nodes）

节点是工作流中的基本执行单元。每个节点是一个函数，它接收当前的状态作为输入，执行一些操作（如调用 LLM、运行工具），然后返回一个更新后的状态（或对状态的修改指令）。

3.3 边（Edges）

边定义了节点之间的流转逻辑。分为两种：

• 条件边（Conditional Edge）：根据当前状态的值（例如，LLM 的输出是继续还是结束）决定下一个要执行的节点。这实现了分支和循环。
• 普通边：无条件地从一个节点指向下一个节点。

3.4 图（Graph）与编译

你将节点和边组合起来，定义一个图结构。然后，通过graph.compile()方法将其编译成一个可执行的、类似链（Chain）的对象。这个编译后的对象管理着状态的传递和节点的调度。

工作流程简述：

1. 初始化一个状态。
2. 将状态传入编译后的图。
3. 图根据当前状态和边逻辑，决定执行哪个节点。
4. 节点执行，更新状态。
5. 重复步骤 3-4，直到到达终止节点。
6. 返回最终状态。

4. 实战：构建你的第一个 LangGraph Agent

让我们从一个经典的“工具调用” Agent开始。这个 Agent 将学会使用一个计算器工具来回答数学问题。

4.1 项目结构初始化

在项目根目录下创建以下文件：

langgraph-agent-tutorial/ ├── basic_agent.py # 基础工具调用Agent ├── memory_agent.py # 带记忆的对话Agent └── requirements.txt # 依赖列表（可由 `pip freeze > requirements.txt` 生成）

4.2 构建基础工具调用 Agent (basic_agent.py)

# basic_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_ollama import ChatOllama from langchain.tools import tool from langchain_core.messages import HumanMessage, AIMessage # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 关键：使用 `operator.add` 让消息列表自动累积 next: str # 用于决定下一个节点 # 2. 创建工具 # 定义一个简单的计算器工具 @tool def calculator(expression: str) -> str: """计算一个数学表达式。支持 +, -, *, /, **。例如：`calculator(\"2 + 3 * 4\")`""" # 警告：实际生产中应对表达式进行严格安全检查，避免代码注入。 try: result = eval(expression) return f"计算结果: {result}" except Exception as e: return f"计算错误: {e}" # 工具列表 tools = [calculator] # 3. 绑定模型和工具 llm = ChatOllama(model="llama3.2:1b", temperature=0) # 为模型绑定工具，使其知道可以调用哪些工具 llm_with_tools = llm.bind_tools(tools) # 4. 定义节点函数 def call_model(state: AgentState): """调用大模型，决定是回复还是调用工具""" print(f"\n[节点 - call_model] 当前消息历史: {state['messages']}") # 获取最新的用户消息 last_message = state['messages'][-1] # 调用绑定了工具的模型 response = llm_with_tools.invoke([last_message]) # 检查模型的响应是普通消息还是工具调用请求 if response.tool_calls: # 模型要求调用工具 print(f" 模型决定调用工具: {response.tool_calls}") # 将模型的响应（包含工具调用信息）添加到消息历史 new_messages = state['messages'] + [response] # 设置下一个节点为 `call_tool` return {"messages": new_messages, "next": "call_tool"} else: # 模型直接给出最终回答 print(f" 模型直接回复: {response.content}") new_messages = state['messages'] + [response] # 设置下一个节点为 `END`，结束流程 return {"messages": new_messages, "next": "__end__"} def call_tool(state: AgentState): """执行模型请求的工具调用""" print(f"\n[节点 - call_tool] 执行工具...") last_message = state['messages'][-1] new_messages = state['messages'].copy() # 遍历模型响应中的所有工具调用请求 for tool_call in last_message.tool_calls: # 根据工具名找到对应的工具函数 tool_to_use = {t.name: t for t in tools}[tool_call['name']] # 执行工具，传入参数 tool_output = tool_to_use.invoke(tool_call['args']) print(f" 执行工具 `{tool_call['name']}`，参数 {tool_call['args']}，结果: {tool_output}") # 将工具执行结果作为一条新消息添加到历史 new_messages.append(AIMessage(content=tool_output, tool_call_id=tool_call['id'])) # 工具执行后，需要再次让模型根据结果进行总结或下一步决策 return {"messages": new_messages, "next": "call_model"} # 5. 构建图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("call_model", call_model) workflow.add_node("call_tool", call_tool) # 设置入口点 workflow.set_entry_point("call_model") # 添加边 workflow.add_conditional_edges( "call_model", # 这是一个路由函数，根据状态中的 `next` 字段决定下一个节点 lambda state: state["next"], # 映射关系：`state["next"]` 的值 -> 下一个节点名 { "call_tool": "call_tool", "__end__": END, } ) workflow.add_edge("call_tool", "call_model") # 工具执行完后无条件回到模型节点 # 编译图 app = workflow.compile() # 6. 运行 Agent if __name__ == "__main__": print("=== 启动基础工具调用 Agent ===") # 初始化状态：用户输入一个问题 initial_state = { "messages": [HumanMessage(content="请问 15 乘以 8 等于多少？")], "next": "call_model" } # 流式输出执行过程（方便观察） print("\n--- 执行流程追踪 ---") for event in app.stream(initial_state, stream_mode="values"): event["messages"][-1].pretty_print() # 获取最终结果 final_state = app.invoke(initial_state) print("\n--- 最终回答 ---") print(final_state["messages"][-1].content)

运行与验证：在终端执行：

python basic_agent.py

你应该能看到类似以下的输出，清晰地展示了 Agent 的思考过程（Reasoning）和行动（Action）：

=== 启动基础工具调用 Agent === --- 执行流程追踪 --- [节点 - call_model] 当前消息历史: [HumanMessage(content='请问 15 乘以 8 等于多少？')] 模型决定调用工具: [{'name': 'calculator', 'args': {'expression': '15 * 8'}, 'id': '...'}] [节点 - call_tool] 执行工具... 执行工具 `calculator`，参数 {'expression': '15 * 8'}，结果: 计算结果: 120 [节点 - call_model] 当前消息历史: [HumanMessage(...), AIMessage(...), AIMessage(content='计算结果: 120', tool_call_id='...')] 模型直接回复: 15 乘以 8 等于 120。 --- 最终回答 --- 15 乘以 8 等于 120。

这个简单的 Agent 已经具备了“思考-行动-观察”的循环能力。模型首先决定调用计算器工具（思考），然后执行工具得到结果（行动-观察），最后根据结果生成面向用户的自然语言回复。

5. 进阶：为 Agent 注入长期记忆（Memory）

没有记忆的 Agent 就像金鱼，每次对话都是新的开始。LangGraph 通过状态管理天然支持记忆。我们将构建一个能记住对话历史的聊天 Agent。

5.1 构建带记忆的对话 Agent (memory_agent.py)

# memory_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage from langgraph.checkpoint.memory import MemorySaver from langgraph.graph.message import add_messages # 1. 定义增强的状态结构，使用 `add_messages` 这个 LangGraph 内置的归约器 class ChatState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] # 自动合并消息列表 user_input: str # 当前轮的用户输入 # 2. 初始化模型和记忆存储 llm = ChatOllama(model="llama3.2:1b", temperature=0.7) # 提高 temperature 使回复更多样 # 创建一个内存检查点存储，用于持久化对话状态（此处为内存，可替换为数据库） memory = MemorySaver() # 3. 定义系统提示词，赋予 Agent 角色和记忆指令 system_prompt = SystemMessage(content="你是一个友好且健谈的助手。请根据我们的对话历史进行自然、连贯的回复。如果用户提到之前聊过的内容，请记得它。") def chat_node(state: ChatState): """聊天节点：结合历史生成回复""" print(f"\n[Chat Node] 历史消息数: {len(state['messages'])}") # 构建给模型的完整消息列表：系统提示 + 历史消息 + 最新用户输入 messages_for_llm = [system_prompt] + state['messages'] + [HumanMessage(content=state['user_input'])] # 调用模型 response = llm.invoke(messages_for_llm) print(f" 助手回复: {response.content}") # 更新状态：将用户输入和AI回复都添加到 messages 中 # `add_messages` 归约器会自动处理合并 return { "messages": [HumanMessage(content=state['user_input']), response], "user_input": "" # 清空当前输入，等待下一轮 } # 4. 构建图（这次更简单，是线性对话流） workflow = StateGraph(ChatState) workflow.add_node("chat", chat_node) workflow.set_entry_point("chat") workflow.add_edge("chat", END) # 单轮对话结束 # 5. 编译图，并注入记忆（检查点）功能 # `checkpointer` 使得每次调用都能基于特定的 `config` 中的线程ID来保存和加载状态 app = workflow.compile(checkpointer=memory) # 6. 运行多轮对话 if __name__ == "__main__": print("=== 启动带记忆的对话 Agent ===") config = {"configurable": {"thread_id": "user_123"}} # 线程ID，标识一个独立的对话会话 # 第一轮对话 print("\n--- 第一轮：自我介绍 ---") initial_state = {"messages": [], "user_input": "你好，我叫小明。"} result = app.invoke(initial_state, config=config) print(f"助手: {result['messages'][-1].content}") # 第二轮对话：测试记忆 print("\n--- 第二轮：询问名字（测试记忆）---") # 注意：我们不需要再传入 `messages`，因为检查点会从 `thread_id` 恢复之前的状态。 # 我们只需要传入新的 `user_input`。 result = app.invoke({"user_input": "你还记得我叫什么名字吗？"}, config=config) print(f"助手: {result['messages'][-1].content}") # 第三轮对话：继续聊天 print("\n--- 第三轮：新话题 ---") result = app.invoke({"user_input": "今天天气怎么样？"}, config=config) print(f"助手: {result['messages'][-1].content}") # 我们可以检查存储的状态 print(f"\n当前对话线程的完整消息历史：") for msg in result['messages']: print(f" {msg.type}: {msg.content[:50]}...")

运行与验证：

python memory_agent.py

输出将展示 Agent 如何记住上下文：

=== 启动带记忆的对话 Agent === --- 第一轮：自我介绍 --- [Chat Node] 历史消息数: 0 助手回复: 你好小明！很高兴认识你。我叫Ollama，是一个AI助手。有什么我可以帮你的吗？ 助手: 你好小明！很高兴认识你。我叫Ollama，是一个AI助手。有什么我可以帮你的吗？ --- 第二轮：询问名字（测试记忆）--- [Chat Node] 历史消息数: 2 # 注意这里历史消息数不再是0，包含了第一轮的对话 助手回复: 当然记得！你刚才告诉我你叫小明。很高兴再次和你聊天，小明！ 助手: 当然记得！你刚才告诉我你叫小明。很高兴再次和你聊天，小明！ --- 第三轮：新话题 --- [Chat Node] 历史消息数: 4 助手回复: 作为一个AI，我无法获取实时天气信息。不过，你可以告诉我你所在的城市，我可以根据一般情况给你一些穿衣或活动建议哦！ 助手: 作为一个AI，我无法获取实时天气信息。不过，你可以告诉我你所在的城市，我可以根据一般情况给你一些穿衣或活动建议哦！

通过MemorySaver和thread_id，我们轻松实现了跨多次invoke调用的长期对话记忆。这是构建实用聊天机器人和复杂会话式 Agent 的基石。

6. 常见问题与排查思路

在开发 LangGraph Agent 过程中，你可能会遇到以下典型问题：

7. 最佳实践与工程建议

将 LangGraph Agent 从原型推向生产，需要考虑以下方面：

7.1 状态设计

• 最小化状态：只将需要在节点间传递和持久化的数据放入状态。避免将整个应用上下文塞进去。
• 使用 Pydantic 模型：对于复杂状态，使用pydantic.BaseModel替代TypedDict，能获得更好的类型验证和序列化支持。
• 清晰的归约器：理解add_messages（用于消息列表）和operator.add（用于普通列表）的区别。自定义归约器可用于复杂合并逻辑。

7.2 节点与图结构

• 节点职责单一：每个节点应只做一件事（如“调用模型”、“调用工具”、“验证输入”）。这提高了可测试性和复用性。
• 利用子图：对于复杂的、可复用的逻辑序列，可以将其封装成一个子图，然后在主图中作为一个节点引用。这有助于管理复杂度。
• 可视化调试：LangGraph 支持导出Mermaid图。在开发阶段，使用workflow.get_graph().draw_mermaid()来可视化你的工作流，检查逻辑是否正确。

7.3 错误处理与鲁棒性

• 节点级 Try-Catch：在节点函数内部对可能失败的操作（如网络调用、工具执行）进行异常捕获，并返回错误信息到状态中，由专门的“错误处理节点”处理。
• 设置超时与中断：对于可能长时间运行或卡住的图，在invoke时设置超时，或利用interrupt_before/after在特定节点前/后设置断点。
• 验证输入：在第一个节点或专门的“验证节点”中对用户输入进行清洗和验证，防止恶意输入或无效数据流入后续流程。

7.4 生产环境部署

• 持久化存储：将MemorySaver替换为RedisSaver或SqliteSaver，以便在服务重启后保留对话状态，并支持多实例部署。
• 异步支持：LangGraph 天然支持异步。对于 I/O 密集型的节点（如调用外部 API），使用async def定义节点函数，并用ainvoke、astream进行调用，可以大幅提升并发性能。
• 配置化管理：将模型参数、工具列表、系统提示词等抽离到配置文件（如config.yaml）或环境变量中，便于不同环境（开发、测试、生产）的切换。
• 日志与监控：为关键节点添加详细的日志记录，记录状态变化、决策路径和耗时。这对于排查生产问题和理解 Agent 行为至关重要。

7.5 测试策略

• 单元测试节点：单独测试每个节点函数，模拟输入状态，断言输出状态。
• 集成测试全图：针对关键用户旅程，编写端到端测试，验证从初始状态到最终输出的正确性。
• 模拟外部依赖：在测试中，使用unittest.mock模拟 LLM 响应和工具调用，使测试快速、稳定且不依赖外部服务。

通过本文的讲解和实战，你已经掌握了使用 LangGraph 和 LangChain 构建具备工具调用和长期记忆能力的 AI 智能体的核心技能。从理解图计算的基本原理，到一步步实现基础 Agent 和记忆 Agent，再到学习生产级的最佳实践，这条路径为你深入探索更复杂的多智能体协作、人工反馈循环等高级主题打下了坚实的基础。建议你接下来尝试修改工具集、设计更复杂的图逻辑，或将其集成到 Web 框架（如 FastAPI）中，打造出真正实用的 AI 应用。