从零构建AI智能体：基于LangGraph的Agent框架实战指南

30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度最近在尝试将大语言模型LLM应用到实际业务场景时发现一个普遍现象很多开发者包括我自己初期都过度关注模型本身——是选 GPT-4、Claude 3 还是某个开源模型参数规模多大推理速度如何然而当我们费尽心思选好模型、调好 API 后却发现模型“智商”在线但“执行力”为零。它无法自动调用工具、无法记忆上下文、无法规划复杂任务更别提自主决策了。这时我才深刻体会到一个强大的 Agent 框架远比一个孤立的“聪明”模型重要得多。本文将从工程实践的角度系统性地拆解 Agent 框架的核心价值、主流架构、实战搭建方法以及避坑指南。无论你是想为现有应用添加智能体能力还是希望构建一个能够自主处理复杂流程的 AI 应用掌握 Agent 框架的设计与实现都是当前 AI 工程化的关键一步。我们将从零开始手把手构建一个具备工具调用、记忆和规划能力的智能体并探讨其背后的设计哲学。1. 背景与核心概念为什么说“框架”比“模型”更重要在讨论 Agent 框架之前我们需要明确几个核心概念。什么是 AI Agent智能体简单来说AI Agent 是一个能够感知环境、自主决策并执行行动以实现特定目标的软件实体。它不仅仅是调用一次大模型 API 获取回答而是一个具备“思考-行动-观察”循环的持续运行系统。一个完整的 Agent 通常包含以下核心组件规划Planning将复杂目标分解为可执行的子任务序列。记忆Memory短期记忆上下文窗口和长期记忆向量数据库等来存储和检索信息。工具使用Tool Use调用外部 API、执行代码、查询数据库等扩展能力。行动Action根据决策执行具体操作。模型 vs. 框架角色定位的差异大语言模型LLM是 Agent 的“大脑”或“推理引擎”。它负责理解指令、进行逻辑推理、生成文本包括代码、计划、思考过程。模型的质量决定了 Agent 的“基础智商”上限。Agent 框架是 Agent 的“神经系统”和“骨骼肌肉”。它定义了 Agent 如何运作的流程工作流、如何管理记忆、如何调用工具、如何处理错误、如何与环境交互。框架的质量决定了 Agent 的“执行力”、“可靠性”和“可扩展性”。一个生动的比喻模型好比一个博学但行动不便的“学者”他知道所有知识但无法自己动手做实验、查资料、写报告。而 Agent 框架则是一个“全能助理”他负责理解学者的意图规划帮学者查阅文献工具调用记录实验过程和结果记忆并最终整理成报告行动。没有框架模型再聪明也只是一个聊天机器人有了框架模型才能成为一个真正能解决问题的智能体。因此标题“Agent框架才是新战场模型不重要”并非指模型毫无价值而是强调在当前的 AI 应用落地阶段工程化、系统化的 Agent 框架能力是释放模型潜力、实现价值闭环的关键瓶颈和竞争焦点。开源模型的快速迭代正在缩小与闭源模型的“智商”差距但如何高效、稳定、安全地构建智能体系统才是开发者们需要深耕的“新战场”。2. 环境准备与版本说明我们将使用 Python 作为开发语言并选择LangChain和LangGraph这两个目前最流行、生态最成熟的 Agent 框架来构建我们的示例。它们提供了从简单链式调用到复杂有状态工作流的一整套工具。基础环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文示例在 macOS/Linux 环境下测试。Python 版本 3.9 推荐 3.10 或 3.11。避免使用 3.12 等较新版本可能遇到依赖兼容性问题。包管理工具pip或conda。核心依赖库及版本我们将创建一个新的虚拟环境来管理依赖。版本号基于 2024 年中期的稳定版本实际开发时请根据官方文档微调。# 1. 创建并激活虚拟环境以 conda 为例 conda create -n ai-agent python3.10 conda activate ai-agent # 2. 安装核心框架 pip install langchain0.1.0 pip install langchain-community0.0.10 pip install langgraph0.0.22 # 3. 安装 OpenAI 模型接口我们将使用其兼容API如DeepSeek、通义千问等 pip install openai1.12.0 # 4. 安装工具调用可能需要的库 pip install requests2.31.0 # 用于网络请求工具 pip install python-dotenv1.0.0 # 用于管理环境变量如API密钥 # 5. 可选如需长期记忆安装向量数据库客户端 # pip install chromadb0.4.22LLM 模型选择与配置为了演示的通用性和可访问性我们将使用DeepSeek的 API因其对国内开发者友好且性价比高。你也可以替换为 OpenAI GPT、Claude、通义千问等任何兼容 OpenAI API 格式的模型。前往 DeepSeek 平台注册并获取 API Key。在项目根目录创建.env文件存放密钥# .env DEEPSEEK_API_KEYyour_deepseek_api_key_here BASE_URLhttps://api.deepseek.com MODELdeepseek-chat在代码中通过环境变量加载配置。项目结构预览ai_agent_demo/ ├── .env # 环境变量配置文件切勿提交至Git ├── requirements.txt # 依赖清单 ├── agent_core.py # Agent核心定义与工具 ├── agent_workflow.py # 基于LangGraph的工作流定义 └── main.py # 主程序入口3. Agent 框架核心原理与主流架构拆解在动手编码前理解主流 Agent 框架背后的设计模式至关重要。这能帮助我们在选择或自研框架时做出正确决策。3.1 ReAct最经典的决策框架搜索材料中提到了ReAct (Reason Act)这是当前绝大多数 Agent 框架的决策核心。其思想是让模型将思考过程Reason和行动步骤Act交织在一起。工作流程思考Thought模型分析当前状态和目标决定下一步该做什么。行动Action模型选择一个工具并生成调用该工具所需的参数。观察Observation框架执行工具调用并将结果成功或失败返回给模型。循环模型基于新的观察再次进行思考直到任务完成或达到终止条件。一个简化的文本表示Thought: 用户想了解北京的天气。我需要调用天气查询工具。 Action: {tool: get_weather, tool_input: {city: 北京}} Observation: 北京今天晴气温15-25°C。 Thought: 我已经获取了天气信息可以给出最终答案了。 Final Answer: 北京今天天气晴朗温度在15到25摄氏度之间。3.2 主流框架类型与选型建议根据搜索材料中提到的“19种主流Agent框架”我们可以将其归纳为几个主要类别框架类型代表项目核心特点适用场景链式调用框架LangChain (早期)将多个LLM调用、工具、记忆按预定顺序链接。结构简单易于理解。简单、线性的任务流程如文档问答、数据提取流水线。有状态工作流框架LangGraph, AutoGen用图Graph来定义Agent的工作流节点是步骤LLM调用、工具边是控制流。支持循环、分支、并行。复杂、多步骤、需状态管理的任务如多轮对话客服、复杂问题拆解、自动化测试。自主智能体框架CrewAI, MetaGPT更高级的抽象支持多智能体协作内置角色分工、任务分配、协调机制。需要团队协作的项目如市场调研报告生成、软件项目开发模拟。轻量级/专用框架LlamaIndex (检索), Haystack专注于某一特定能力如检索增强生成RAG。需要与特定数据源深度集成的场景。选型建议新手入门/快速验证从 LangChain 的简单 Agent 开始。复杂业务逻辑/生产环境强烈推荐 LangGraph。它将 Agent 的状态和流程显式化调试和监控更方便是构建可靠 Agent 系统的未来方向。模拟团队协作研究 CrewAI 或 MetaGPT。接下来我们将重点使用LangGraph来构建一个具备完整 ReAct 循环的智能体。4. 完整实战构建一个多功能 AI 智能体我们的目标是构建一个能处理以下任务的智能体回答一般性知识问题。查询实时信息如天气。进行简单的计算。在对话中记住上下文。4.1 定义工具Tool工具是 Agent 能力的延伸。我们先定义三个基础工具。# agent_core.py import os import requests import json from datetime import datetime from typing import TypedDict, Optional from dotenv import load_dotenv from langchain.tools import tool from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI # 加载环境变量 load_dotenv() # 初始化LLM (使用DeepSeek兼容OpenAI API) llm ChatOpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(BASE_URL), modelos.getenv(MODEL), temperature0.1, # 降低随机性使Agent更稳定 ) # 工具1: 获取天气模拟 tool def get_weather(city: str) - str: 获取指定城市的当前天气信息。 Args: city: 城市名称例如“北京”、“上海”。 Returns: 该城市的天气情况描述字符串。 # 注意这里是一个模拟函数。真实场景应调用如和风天气、OpenWeatherMap等API。 # 为安全起见不在此处嵌入真实API调用代码。 weather_map { 北京: 晴朗15-25°C微风, 上海: 多云18-28°C东南风3级, 深圳: 阵雨22-30°C南风4级, } return weather_map.get(city, f抱歉未找到{city}的天气信息。模拟数据仅支持北京、上海、深圳。) # 工具2: 计算器 tool def calculator(expression: str) - str: 执行一个简单的数学表达式计算。支持加减乘除和括号。 Args: expression: 数学表达式例如“(12 5) * 3”。 Returns: 计算结果字符串。 # 警告在生产环境中直接使用eval是危险的容易造成代码注入。 # 此处仅用于演示。实际应用应使用安全库如ast.literal_eval或专门的计算库。 try: # 极简的安全检查只允许数字、运算符和括号 allowed_chars set(0123456789-*/(). ) if not all(c in allowed_chars for c in expression): return 错误表达式包含非法字符。 result eval(expression) return f{expression} {result} except Exception as e: return f计算错误{e} # 工具3: 获取当前时间 tool def get_current_time(timezone: Optional[str] None) - str: 获取当前日期和时间。 Args: timezone: 时区可选例如“Asia/Shanghai”。默认为系统时区。 Returns: 格式化的日期时间字符串。 now datetime.now() # 简化处理实际应用可使用pytz库 if timezone and shanghai in timezone.lower(): # 模拟东八区 from datetime import timedelta now timedelta(hours8) return now.strftime(%Y-%m-%d %H:%M:%S) # 将工具打包成列表供Agent使用 tools [get_weather, calculator, get_current_time] # 为LLM绑定工具使其知道有哪些工具可用 llm_with_tools llm.bind_tools(tools)4.2 定义 Agent 状态与工作流使用 LangGraphLangGraph 的核心是定义状态State和组成工作流的节点Nodes。# agent_workflow.py from typing import TypedDict, Annotated, Sequence import operator from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage from agent_core import llm_with_tools, tools # 1. 定义状态结构 class AgentState(TypedDict): Agent运行时的状态。 messages: 完整的对话消息历史包括用户输入、AI回复、工具调用及结果。 messages: Annotated[Sequence[BaseMessage], operator.add] # operator.add表示该字段会累积 # 2. 创建工具执行器 tool_executor ToolExecutor(tools) # 3. 定义工作流中的节点 def call_model(state: AgentState): 调用大语言模型节点。 输入当前状态包含历史消息。 输出更新后的状态添加了模型的回复消息。 messages state[messages] # 调用已绑定工具的LLM response llm_with_tools.invoke(messages) # 将模型的响应添加到消息历史中 return {messages: [response]} def execute_tools(state: AgentState): 执行工具调用节点。 输入状态最后一条消息应是AIMessage且包含工具调用请求。 输出更新后的状态添加了工具执行结果的消息。 messages state[messages] last_message messages[-1] # 初始化结果列表 tool_messages [] # 遍历模型请求调用的所有工具 for tool_call in last_message.tool_calls: # 执行工具 result tool_executor.invoke(tool_call) # 构造ToolMessage这是LangChain约定的工具结果消息格式 tool_messages.append(ToolMessage( contentstr(result), tool_call_idtool_call[id], # 必须与请求的id对应 nametool_call[name] )) return {messages: tool_messages} # 4. 定义条件路由逻辑 def should_continue(state: AgentState) - str: 根据模型的最新回复决定下一步是继续调用工具还是结束。 这是一个路由函数返回下一个节点的名称。 messages state[messages] last_message messages[-1] # 如果模型的最新回复中没有工具调用说明任务完成可以结束。 if not last_message.tool_calls: return end # 否则还需要继续执行工具。 return continue # 5. 构建工作流图 def create_agent_workflow(): 创建并编译一个ReAct风格的Agent工作流图。 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, call_model) # “思考”节点 workflow.add_node(action, execute_tools) # “行动”节点 # 设置入口点 workflow.set_entry_point(agent) # 添加边定义流程 workflow.add_conditional_edges( agent, # 从哪个节点出发 should_continue, # 路由判断函数 { continue: action, # 如果需要继续去执行工具 end: END # 如果结束直接到终点 } ) workflow.add_edge(action, agent) # 执行完工具后回到agent继续思考 # 编译图 return workflow.compile() # 创建图实例 graph create_agent_workflow()4.3 创建主程序并运行测试现在我们将所有部分组合起来并运行几个测试用例。# main.py from agent_workflow import graph from langchain_core.messages import HumanMessage import asyncio async def run_agent_async(query: str): 异步运行Agent处理一次用户查询。 # 初始化状态包含用户输入 initial_state {messages: [HumanMessage(contentquery)]} print(f\n{*50}) print(f用户输入: {query}) print(f{*50}) # 用于存储最终结果 final_response None # 流式执行图并实时打印每一步 async for event in graph.astream(initial_state): for key, value in event.items(): if key agent: # 打印模型的“思考”和“行动”决定 msg value.get(messages, [])[-1] if msg.tool_calls: print(f[Agent 思考] 决定调用工具:) for tc in msg.tool_calls: print(f - 工具: {tc[name]}, 参数: {tc[args]}) else: print(f[Agent 最终回答]: {msg.content}) final_response msg.content elif key action: # 打印工具执行结果 for msg in value.get(messages, []): print(f[工具执行结果]: {msg.content}) print(-*30) return final_response async def main(): 主测试函数 test_queries [ 北京今天的天气怎么样, 请计算一下 (15 7) * 3 等于多少, 现在是什么时间, 先告诉我上海的天气然后计算北京气温假设25度和深圳气温假设30度的平均值。, ] for query in test_queries: await run_agent_async(query) await asyncio.sleep(1) # 简单间隔避免请求过快 if __name__ __main__: asyncio.run(main())4.4 运行结果与说明运行python main.py你将看到类似以下的输出具体内容因模型回复而异 用户输入: 北京今天的天气怎么样 [Agent 思考] 决定调用工具: - 工具: get_weather, 参数: {city: 北京} ------------------------------ [工具执行结果]: 晴朗15-25°C微风 ------------------------------ [Agent 最终回答]: 北京今天的天气是晴朗气温在15到25摄氏度之间有微风。 ------------------------------ 用户输入: 先告诉我上海的天气然后计算北京气温假设25度和深圳气温假设30度的平均值。 [Agent 思考] 决定调用工具: - 工具: get_weather, 参数: {city: 上海} ------------------------------ [工具执行结果]: 多云18-28°C东南风3级 ------------------------------ [Agent 思考] 决定调用工具: - 工具: calculator, 参数: {expression: (25 30) / 2} ------------------------------ [工具执行结果]: (25 30) / 2 27.5 ------------------------------ [Agent 最终回答]: 上海的天气是多云18-28°C东南风3级。北京25度和深圳30度的平均气温是27.5度。 ------------------------------代码解读工作流循环对于复杂查询Agent 展示了完整的 ReAct 循环。它先思考需要调用天气工具执行后获得观察结果再思考需要调用计算器工具最后综合所有信息给出最终答案。状态管理AgentState中的messages列表完整记录了整个对话和工具调用的历史这既是短期记忆也是传递给模型的下文。图结构agent - (判断) - action - agent ... - end的循环清晰可见。LangGraph将这一控制流可视化、模块化。5. 进阶为 Agent 添加记忆与更复杂的规划基础的 ReAct Agent 已经能处理许多任务。但对于真实应用我们还需要长期记忆和更高级的规划能力。5.1 集成向量数据库实现长期记忆让 Agent 记住跨对话的信息。这里以 ChromaDB 为例。# memory_integration.py (扩展示例) from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain_core.documents import Document from langchain.chains import create_history_aware_retriever, create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder import os # 初始化嵌入模型和向量库 embeddings OpenAIEmbeddings( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(BASE_URL), modeltext-embedding-3-small, # 检查DeepSeek是否提供嵌入模型或使用其他兼容模型 ) vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) # 假设我们有一些历史文档或对话片段需要记忆 historical_data [ 项目A的负责人是张三预计下周交付。, 客户的偏好是喜欢简洁的UI设计。, 系统登录API的地址是 /api/v1/login。 ] # 将数据存入向量库实际应用中可能是增量添加 vectorstore.add_documents([Document(page_contenttext) for text in historical_data]) # 创建一个“历史感知”的检索器 retriever vectorstore.as_retriever() prompt ChatPromptTemplate.from_messages([ MessagesPlaceholder(variable_namechat_history), (user, {input}), (user, 根据以上对话生成一个搜索查询来查找相关信息。) ]) history_aware_retriever create_history_aware_retriever(llm, retriever, prompt) # 然后可以将这个检索器作为一个“工具”集成到之前的Agent中 # 或者构建一个专门的“检索链”来处理需要记忆的查询5.2 实现分层任务规划Planning对于“写一份周报”这样的复杂指令Agent 需要先规划子任务。我们可以通过一个专门的“规划器”节点来实现。# planning_node.py (概念示例) from langchain_core.prompts import PromptTemplate PLANNER_PROMPT PromptTemplate.from_template( 你是一个任务规划专家。请将以下复杂任务分解为一个清晰的、顺序执行的子任务列表。 每个子任务应该是一个具体的、可执行的指令可以被一个具备工具调用能力的AI助手理解。 只输出JSON格式的列表例如[子任务1描述, 子任务2描述, ...] 复杂任务{task} 子任务列表 ) def planning_node(state: AgentState): 规划节点将复杂任务分解。 这是一个简化示例实际规划器可能更复杂并能与工具列表结合。 task state[messages][-1].content # 获取最新用户任务 planner_chain PLANNER_PROMPT | llm # 创建简单链 plan_json_str planner_chain.invoke({task: task}).content # 解析JSON这里简化处理 import json try: sub_tasks json.loads(plan_json_str.strip()) except: sub_tasks [f执行任务: {task}] # 解析失败则退回原任务 # 将规划结果以特定格式存入状态供后续节点使用 # 例如可以添加一个 plan 字段到 state 中 # 这里我们简单地将第一个子任务作为新消息发出 new_message HumanMessage(contentsub_tasks[0]) return {messages: [new_message]} # 然后你需要修改工作流图在 agent 节点前加入 planning 节点并根据任务复杂度决定是否走规划分支。6. 常见问题与排查思路在开发和使用 Agent 框架时你会遇到一些典型问题。问题现象常见原因解决思路Agent 陷入死循环不停调用工具1. 工具返回结果未满足终止条件。2. LLM 未能正确理解任务已完成。3. 状态管理错误历史消息堆积导致混乱。1. 检查工具输出格式是否清晰、完整。2. 优化系统提示词System Prompt明确告知“何时停止”。3. 在should_continue函数中添加更严格的终止逻辑如最大循环次数。4. 使用 LangGraph 的检查点Checkpoint功能进行调试。LLM 不调用工具直接回答1. 工具描述不清晰。2. 提示词未有效激发工具使用。3. 模型本身工具调用能力弱特别是小模型。1. 为每个工具编写清晰、具体的文档字符串Docstring。2. 在系统提示词中强调“你必须使用工具”。3. 在bind_tools时提供工具调用格式的少样本示例few-shot examples。4. 考虑升级模型或使用专门优化过工具调用的模型。工具调用参数错误或格式不对1. LLM 生成的参数 JSON 格式错误。2. 参数类型与工具函数声明不匹配。3. 工具函数本身有异常。1. 使用Pydantic为工具参数定义严格的类型和验证。2. 在工具执行层添加健壮的异常处理和类型转换。3. 在开发阶段打印出模型生成的原始tool_calls进行调试。处理长上下文时性能下降或丢失信息1. 上下文窗口有限历史消息过长。2. 无关信息干扰模型决策。1. 实现记忆摘要定期让模型总结对话历史用摘要替换冗长的原始消息。2. 使用向量检索记忆只将与当前查询最相关的历史片段放入上下文。3. 设计状态过滤在状态中只保留必要信息而非全部消息。多 Agent 协作时通信混乱1. 角色和职责定义不清。2. 缺乏统一的协调机制或仲裁者。1. 为每个 Agent 定义清晰的系统提示词明确其角色和权限。2. 使用CrewAI或LangGraph的StateGraph明确编排多 Agent 的工作流和消息传递路径。3. 引入一个“主管”Agent 来分配任务和汇总结果。7. 最佳实践与工程建议要将 Agent 从 demo 推向生产必须关注以下工程实践。1. 提示词工程是核心系统提示词System Prompt必须清晰定义 Agent 的角色、目标、可用工具、输出格式和行动准则。这是 Agent 的“宪法”。少样本示例Few-Shot在提示词中提供 1-2 个完整的 ReAct 循环示例能极大提升模型工具调用的准确性和格式规范性。结构化输出要求模型以 JSON 等固定格式输出思考和行动便于后端解析。2. 工具设计原则单一职责一个工具只做一件事。search_web和calculate应该分开。健壮性工具函数内部必须有完善的错误处理try-except返回对模型友好的错误信息而不是 Python 异常栈。安全性永远不要相信来自 LLM 的输入直接执行。像calculator工具中的eval是极度危险的示例。生产环境必须使用沙箱、白名单或专用库。文档清晰工具的docstring是模型理解如何调用它的主要依据务必详细描述功能、参数和返回值。3. 状态与流程管理显式状态像 LangGraph 一样使用显式的、类型化的状态对象。避免使用全局变量或隐式上下文。可观测性在关键节点调用模型、调用工具、状态转换记录日志和指标。这比调试黑盒 Agent 容易得多。持久化与检查点对于长时运行的任务支持将 Agent 状态保存到数据库并能从中断点恢复。4. 评估与测试单元测试工具确保每个工具函数在各种边界条件下都能正确工作。集成测试工作流用一组固定的输入测试整个 Agent 图确保输出符合预期。评估指标定义成功指标如任务完成率、工具调用准确率、平均步数等。使用LangSmith等平台进行跟踪和评估。5. 安全与合规权限控制为不同工具设置权限等级。例如查询数据库的工具比发送邮件的工具权限更低。用户确认对于高风险操作如删除数据、发送邮件设计“人工确认”环节或让 Agent 生成待用户确认的草稿。输入输出过滤对用户输入和模型输出进行内容安全过滤防止注入攻击或生成有害内容。构建 AI Agent 系统是一场关于工程严谨性与创造力的平衡。模型提供了理解世界的潜力而框架则是将这种潜力转化为可靠、安全、有价值行动的铁轨。从理解 ReAct 模式开始选择一个像 LangGraph 这样健壮的框架遵循最佳实践从小而具体的功能迭代你就能在这个“新战场”上构建出真正解决问题的智能应用。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度