1. 项目概述为什么现在人人都想搭一个Agent最近和不少同行交流发现一个挺有意思的现象无论是做后端开发、数据分析还是做产品经理大家聊天的话题总绕不开“Agent”智能体。好像一夜之间不会搭个Agent框架就跟不上技术潮流了。这股热潮背后其实反映的是我们对现有AI应用方式的一种“不满足”。过去我们调用大模型API更像是“一问一答”的客服模式我们得把问题拆解得极其细致模型才能给出靠谱答案。但现实世界的问题往往是复杂、多步骤的比如“帮我分析一下上季度的销售数据找出问题并生成一份给老板的汇报PPT”。这种需求靠单次API调用几乎不可能完成。这就是Agent框架要解决的核心问题让AI具备自主规划、使用工具、持续执行并反思调整的能力从而完成一个复杂的、多步骤的目标。你可以把它想象成给大模型配了一个“私人助理”和一套“工具箱”。这个助理Agent核心负责理解你的终极意图拆解任务制定计划工具箱Tools里有各种专业工具比如搜索、计算、读写文件、调用特定API而框架本身就是一套管理它们协同工作的规则和流程。所以“从零搭建Agent框架”这个事听起来高大上但内核很实在。它不是为了炫技而是为了真正把大模型的潜力“工程化”、“实用化”。无论是想做个能自动处理工单的客服助手还是想开发一个能联网搜索、分析信息并撰写报告的研究助理抑或是想构建一个能根据用户描述自动调整参数的智能设计工具一个灵活、可控的自主Agent框架都是基石。接下来我就结合自己趟过的坑手把手带你走一遍从设计到实现的完整路径。2. 核心设计思路你的Agent大脑里应该有什么在动手写代码之前我们必须想清楚框架的“灵魂”。一个健壮的Agent框架绝不是把几个开源库拼在一起就完事了。它需要一套清晰、可扩展的架构设计。经过多次迭代我认为一个最小可行且易于扩展的Agent框架核心应包含以下四个模块它们共同构成了Agent的“大脑”2.1 思维链与规划器任务拆解的“指挥官”这是Agent的“思考”核心。它的职责是理解用户输入的最终目标Goal并将其分解成一系列可执行的子任务Sub-tasks。这里的关键是“链式思考”Chain-of-Thought和“规划”Planning。为什么需要规划器直接让大模型输出最终答案对于复杂任务来说失败率极高。规划器迫使模型进行“逐步推理”。例如目标“写一份关于新能源汽车的市场分析报告”。一个简单的规划器可能会引导模型产出如下步骤搜索并收集近期新能源汽车行业的政策、销量数据、头部公司动态。对收集到的信息进行归纳整理提炼出核心趋势、挑战与机遇。根据分析结果搭建报告框架引言、市场现状、竞争分析、未来展望等。撰写报告正文。检查并润色报告。如何实现最简单的实现方式是设计一个特定的“系统提示词”System Prompt引导大模型扮演“规划者”角色。你可以这样设计提示词“你是一个任务规划专家。请将用户给出的复杂目标分解成一个有序的、可操作的任务列表。每个任务应该足够具体使得一个执行单元例如一个工具调用能够完成它。请以JSON格式输出包含tasks数组每个任务有id,description,expected_output字段。”更高级的规划器可以引入“反思”机制即在执行完某些步骤后重新评估剩余计划是否需要调整。实操心得规划器的粒度把控规划器拆解任务的“粒度”是个艺术。太粗如“完成市场分析”执行器无从下手太细如“打开浏览器在搜索框输入‘新能源汽车 政策’…”会极大增加冗余和出错概率。我的经验是一个子任务最好对应一个“工具”的一次调用或一段连贯的“推理”。比如“搜索新能源汽车政策”对应搜索工具“分析数据趋势”对应代码解释器工具。在实践中需要根据任务领域和可用工具集反复调整提示词来校准这个粒度。2.2 工具集Agent的“瑞士军刀”工具Tools是Agent与外部世界交互的桥梁。没有工具的Agent只是一个“思想家”有了工具才成为“行动派”。框架需要一套标准化的方式来定义、管理和调用工具。工具的定义一个工具通常包含名称Name唯一标识如web_search,python_executor。描述Description清晰说明工具的功能这个描述会被送入大模型帮助它决定何时调用该工具。描述的质量直接决定工具调用的准确性。参数模式Arguments Schema定义工具需要的输入参数通常用JSON Schema描述。执行函数Function实际的代码实现。工具集的管理框架需要维护一个“工具注册表”。当规划器生成一个子任务时Agent核心需要根据任务描述从注册表中匹配合适的工具。这里通常利用大模型的函数调用Function Calling能力。你将所有工具的描述和参数模式以特定格式如OpenAI的tools格式传给模型模型会返回它认为应该调用哪个工具以及参数是什么。常见工具类型信息获取类网络搜索如SerpAPI、DuckDuckGo、数据库查询、API调用如天气、股票。计算与处理类Python代码执行器处理数据、计算、图像处理、文本摘要。输出与操作类文件读写txt, csv, pdf、发送邮件、生成图表。2.3 记忆与状态管理记住“刚才发生了什么”Agent在执行多步骤任务时必须有“记忆”。记忆分为短期和长期短期记忆/对话历史记录当前任务执行过程中的所有交互包括用户输入、Agent的思考、工具调用及结果。这是进行连贯推理的基础。长期记忆/知识库存储跨会话的持久化信息例如用户偏好、历史任务总结的经验。这可以通过向量数据库如Chroma, Pinecone来实现。状态管理的关键框架需要维护一个“执行状态”对象。它至少包括最终目标Goal用户最初提出的任务。当前计划Plan由规划器生成的任务列表。执行进度Progress哪些任务已完成结果是什么当前正在执行哪个任务。上下文Context整合了相关的对话历史、工具执行结果作为下一次模型调用的输入。一个良好的状态管理机制能确保Agent在被打断或执行出错后仍能理解当前处境并做出合理决策。2.4 执行引擎与调度器让一切运转起来的“心脏”这是框架的驱动模块。它负责串联整个流程接收目标获取用户输入。初始化规划调用规划器生成任务列表初始化状态。循环执行 a.任务选择根据状态选择下一个待执行的任务。 b.推理与工具调用将当前任务描述、历史上下文、可用工具列表传给大模型。模型决定是进行“思考”输出文本还是“行动”调用工具。 c.执行工具如果模型决定调用工具调度器则找到对应工具函数传入参数并执行。 d.结果处理将工具执行结果或模型思考内容更新到状态和记忆中。 e.状态检查判断当前任务是否完成是否所有任务已完成或是否出现错误需要调整计划。输出最终结果所有任务完成后整合信息生成最终输出给用户。调度器还需要处理错误和异常比如工具调用失败、模型返回格式错误等并设计重试或回退策略。3. 从零开始一步步实现你的第一个Agent框架理论说得再多不如动手一行代码。我们使用Python来构建因为它有最丰富的AI生态。假设我们的目标是构建一个能联网搜索并总结信息的Agent。我们选择OpenAI API或其他兼容API的大模型作为“大脑”LangChain作为辅助工具库但核心逻辑我们自己掌控以便理解原理。3.1 环境准备与依赖安装首先创建一个干净的Python环境推荐使用conda或venv然后安装核心依赖。# 创建并激活虚拟环境以venv为例 python -m venv agent_env source agent_env/bin/activate # Linux/Mac # agent_env\Scripts\activate # Windows # 安装核心库 pip install openai langchain langchain-community # 安装用于网络搜索的工具库例如duckduckgo-search pip install duckduckgo-search # 安装用于结构化输出的库方便解析模型返回的工具调用 pip install pydantic这里解释一下选型openai官方SDK用于调用GPT系列模型。你也可以替换为其他兼容OpenAI API格式的库如litellm。langchain我们主要利用其丰富的工具集成和便捷的提示模板管理但Agent的核心循环我们自行编写以保证灵活性和学习目的。duckduckgo-search一个免费、无需API key的搜索工具包适合演示。生产环境可以考虑SerpAPI等更稳定的服务。3.2 定义核心组件工具、记忆与状态我们从一个简单的文件开始比如agent_core.py。第一步定义工具基类和具体工具我们设计一个简单的工具抽象所有工具都继承它。from abc import ABC, abstractmethod from typing import Any, Dict import json from duckduckgo_search import DDGS class BaseTool(ABC): 工具基类 name: str description: str abstractmethod def _run(self, **kwargs) - str: 工具的执行逻辑 pass def run(self, tool_input: str) - str: 对外统一的运行接口解析输入字符串为参数 try: args json.loads(tool_input) except json.JSONDecodeError: args {query: tool_input} # 简化处理假设单参数 return self._run(**args) def to_function_schema(self) - Dict: 生成用于模型函数调用的模式 # 这是一个简化版本实际需要根据工具参数动态生成 return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: { query: {type: string, description: 搜索查询词} }, required: [query] } } } class WebSearchTool(BaseTool): 网络搜索工具 def __init__(self): self.name web_search self.description 使用DuckDuckGo在互联网上搜索最新信息。输入应为搜索关键词。 def _run(self, query: str, max_results: int 5) - str: try: with DDGS() as ddgs: results [r for r in ddgs.text(query, max_resultsmax_results)] # 格式化结果 formatted \n---\n.join([f标题{r[title]}\n链接{r[href]}\n摘要{r[body]} for r in results]) return f搜索 {query} 的结果\n{formatted} except Exception as e: return f搜索工具执行出错{str(e)} class CalculatorTool(BaseTool): 计算器工具使用Python eval生产环境需沙箱隔离 def __init__(self): self.name calculator self.description 执行数学计算。输入为一个数学表达式字符串如 3 5 * 2。 def _run(self, expression: str) - str: try: # 警告直接eval有安全风险仅用于演示。生产环境必须使用沙箱如restrictedpython或数学解析库。 result eval(expression, {__builtins__: {}}, {}) return f计算结果{expression} {result} except Exception as e: return f计算错误{str(e)}重要警告关于计算器工具的安全上面CalculatorTool使用了eval()这在演示中简单但在任何面向用户的生产环境中都是极度危险的因为它允许执行任意Python代码。必须替换为安全的方案例如使用ast.literal_eval()处理纯数字和运算符。使用专门的数学表达式解析库如numexpr。在严格的沙箱环境如restrictedpython中执行。在你的项目中请务必采用安全方案。第二步定义记忆与状态from dataclasses import dataclass, field from typing import List, Dict, Any dataclass class AgentState: Agent执行状态 goal: str # 最终目标 plan: List[Dict] field(default_factorylist) # 任务计划每个任务是一个dict completed_tasks: List[Dict] field(default_factorylist) # 已完成的任务及结果 current_task_index: int 0 # 当前执行到第几个任务 context: str # 当前累积的上下文信息 max_steps: int 20 # 最大执行步数防止死循环 def update_context(self, new_info: str): 更新上下文并保持一定长度避免token超限 self.context f\n{new_info} # 简单的截断策略保留最近N个字符 if len(self.context) 4000: self.context self.context[-4000:] def is_finished(self) - bool: 判断任务是否完成 return self.current_task_index len(self.plan) and len(self.plan) 0 def get_current_task(self) - Dict: 获取当前任务 if self.current_task_index len(self.plan): return self.plan[self.current_task_index] return None3.3 构建核心执行引擎这是最核心的部分我们创建一个AgentEngine类。import openai from typing import List, Optional import json import re class AgentEngine: def __init__(self, api_key: str, model: str gpt-3.5-turbo, tools: Optional[List[BaseTool]] None): self.client openai.OpenAI(api_keyapi_key) self.model model self.tools tools or [] self.tool_map {tool.name: tool for tool in self.tools} def _call_llm(self, messages: List[Dict], tools: Optional[List[Dict]] None) - Dict: 调用大模型支持函数调用 params { model: self.model, messages: messages, temperature: 0.1, # 低温度保证决策稳定性 } if tools: params[tools] tools params[tool_choice] auto # 让模型自行决定是否调用工具 response self.client.chat.completions.create(**params) return response.choices[0].message def create_plan(self, goal: str) - List[Dict]: 规划器将目标拆解为任务列表 system_prompt 你是一个顶尖的任务规划专家。请将用户给出的复杂目标分解成一个有序的、可操作的任务列表。 每个任务应该足够具体使得一个执行单元例如一个工具调用能够完成它。 请以严格的JSON格式输出只包含一个 tasks 键其值是一个数组。 数组中的每个元素是一个对象包含 id从1开始的序号, description任务描述, expected_output期望产出三个字段。 示例 目标“查一下特斯拉最近的股价并计算如果买入100股需要多少钱” 输出 { tasks: [ {id: 1, description: 搜索特斯拉(TSLA)的最新股价, expected_output: 特斯拉当前股价美元}, {id: 2, description: 计算购买100股特斯拉股票所需的总金额, expected_output: 总金额美元} ] } messages [ {role: system, content: system_prompt}, {role: user, content: f目标{goal}} ] response self._call_llm(messages) content response.content # 尝试从响应中解析JSON有时模型会在JSON外加 Markdown 代码块或说明文字 try: # 尝试匹配 JSON 部分 json_match re.search(r\{.*\}, content, re.DOTALL) if json_match: plan_data json.loads(json_match.group()) else: plan_data json.loads(content) return plan_data.get(tasks, []) except json.JSONDecodeError as e: print(f规划器返回无法解析的JSON: {content}) # 降级方案返回一个简单的默认计划 return [{id: 1, description: goal, expected_output: 完成目标}] def execute_step(self, state: AgentState) - (str, bool): 执行单一步骤思考或行动 current_task state.get_current_task() if not current_task: return 没有更多任务需要执行。, True # 构建给模型的提示 user_prompt f 当前总体目标{state.goal} 当前待执行任务{current_task[description]} (期望产出{current_task[expected_output]}) 以下是到目前为止的上下文记录 {state.context} 请你根据以上信息思考如何完成当前任务。你可以选择 1. 直接输出你的思考或答案如果任务仅需推理。 2. 调用合适的工具来获取信息或执行操作。 请做出你的决策。 messages [ {role: system, content: 你是一个善于使用工具解决问题的助手。请根据任务需求决定是直接回答还是调用工具。}, {role: user, content: user_prompt} ] # 准备工具列表函数调用格式 tools_for_llm [tool.to_function_schema() for tool in self.tools] response_message self._call_llm(messages, toolstools_for_llm) tool_calls response_message.tool_calls if hasattr(response_message, tool_calls) else None if tool_calls: # 模型决定调用工具 for tool_call in tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) if tool_name in self.tool_map: tool self.tool_map[tool_name] print(f[Agent] 调用工具: {tool_name}, 参数: {tool_args}) tool_result tool.run(json.dumps(tool_args)) state.update_context(f调用工具 {tool_name} 结果{tool_result}) # 将工具结果作为模型的后续输入让模型进行总结或下一步决策简化处理此处直接记录 # 更完整的实现应在此发起一个新的LLM调用让模型基于工具结果继续 return f已使用工具[{tool_name}]完成任务结果已记录。, False else: error_msg f尝试调用未知工具{tool_name} state.update_context(f错误{error_msg}) return error_msg, False else: # 模型直接输出思考结果 llm_output response_message.content state.update_context(f模型推理{llm_output}) print(f[Agent] 模型思考{llm_output[:100]}...) # 检查模型输出是否暗示任务完成这里逻辑可以更智能 if 完成 in llm_output or 综上 in llm_output or len(llm_output) 50: # 假设模型给出了一个完整的答案认为当前任务完成 return llm_output, True else: # 模型可能还在思考中需要继续 return llm_output, False def run(self, goal: str) - str: 主执行循环 print(f[开始] 目标{goal}) # 1. 规划 tasks self.create_plan(goal) print(f[规划] 生成任务列表{tasks}) state AgentState(goalgoal, plantasks) # 2. 执行循环 step_count 0 final_result while not state.is_finished() and step_count state.max_steps: step_count 1 print(f\n[步骤 {step_count}] 执行任务 {state.current_task_index 1}/{len(tasks)}) step_output, task_completed self.execute_step(state) if task_completed: # 当前任务完成记录结果移动到下一个任务 if state.get_current_task(): state.completed_tasks.append({ task: state.get_current_task(), result: step_output }) state.current_task_index 1 print(f[完成] 任务 {state.current_task_index} 完成。) # 否则任务未完成下一轮循环继续处理同一任务例如需要多次工具调用 final_result step_output # 暂存最后一步输出 # 3. 汇总 if state.is_finished(): print(f\n[成功] 所有任务执行完毕) # 简单汇总这里可以添加一个“总结器”步骤让模型基于所有上下文生成最终答案 summary_prompt f基于以下所有执行记录请给用户一个关于目标 {goal} 的完整、简洁的最终答复。\n上下文{state.context} summary_msg self._call_llm([{role: user, content: summary_prompt}]) return summary_msg.content else: print(f\n[中断] 达到最大步数限制或执行出错。) return f任务执行未完全完成。最后状态{final_result}\n上下文{state.context[:500]}...3.4 组装并运行你的第一个Agent最后我们创建一个主程序来启动一切。# main.py import os from agent_core import AgentEngine, WebSearchTool, CalculatorTool def main(): # 从环境变量读取API Key确保安全 api_key os.getenv(OPENAI_API_KEY) if not api_key: print(错误请设置 OPENAI_API_KEY 环境变量。) return # 1. 初始化工具集 tools [ WebSearchTool(), # CalculatorTool(), // 注意演示中暂不启用因为安全警告 ] # 2. 创建Agent引擎 agent AgentEngine(api_keyapi_key, modelgpt-3.5-turbo, toolstools) # 3. 运行一个目标 goal 查一下苹果公司Apple Inc.最新发布的产品是什么并简要总结其主要特点。 final_answer agent.run(goal) print(\n *50) print(【最终答案】) print(final_answer) print(*50) if __name__ __main__: main()运行这个程序你会看到控制台输出Agent的思考过程先规划任务然后调用搜索工具获取信息最后整合输出答案。恭喜你你已经拥有了一个最基础但五脏俱全的自主Agent框架4. 避坑指南与进阶优化第一次运行很可能不会一帆风顺。下面是我在实践中总结的常见问题和进阶优化方向。4.1 常见问题与排查技巧问题1模型不调用工具总是自言自语。可能原因1工具描述不清晰。模型无法理解工具能干什么。解决优化工具的描述description字段使其更精准并与任务描述对齐。例如“搜索网络信息”就比“搜索工具”好得多。可能原因2提示词未引导。系统提示词没有明确鼓励或指导模型使用工具。解决在系统提示词中强调“你是一个善于使用工具的助手”并在用户提示词中明确给出选项如“你可以选择直接回答或调用工具”。可能原因3任务过于简单。模型认为无需工具即可回答。解决测试时使用明确需要外部信息的任务如“今天北京的天气如何”问题2模型调用工具时参数错误。可能原因参数模式Schema与工具实际输入不匹配。解决确保to_function_schema方法返回的parameters定义与工具_run方法的参数名和类型完全一致。使用Pydantic模型来定义Schema可以大大减少错误。问题3陷入死循环或步骤混乱。可能原因状态管理逻辑有缺陷。例如任务完成条件判断不准。解决引入更明确的任务完成标志。可以让模型在输出中明确声明“任务完成”。设置最大步数max_steps作为安全阀。实现“反思”步骤每执行几步后让模型回顾一下进度和计划是否需要调整。问题4Token消耗巨大成本高。可能原因上下文context无限增长。解决压缩历史不是存储完整的对话记录而是定期让模型总结之前的交互用总结代替详细历史。选择性记忆只保留与当前任务最相关的历史片段。可以利用向量数据库检索相关记忆。使用更经济的模型规划、工具选择等步骤可以使用小模型如GPT-3.5最终汇总答案再用大模型如GPT-4。4.2 从Demo到生产关键优化点上面的框架是一个教学原型。要用于实际项目你需要考虑以下方面1. 更强大的规划与反思ReAct, Plan-and-Execute实现ReActReasoning Acting模式让模型在每一步都输出“思考Thought”、“行动Action”、“观察Observation”的循环。这能让推理过程更透明、更稳定。引入层级任务分解Hierrachical Task Decomposition对于极其复杂的任务可以进行多级分解。在任务失败或结果不理想时触发反思Reflection步骤让模型分析原因并调整计划。2. 工具管理的增强动态工具加载根据任务类型动态加载不同的工具集减少不必要的干扰。工具组合Tool Composition设计能让Agent自动组合使用多个工具完成复杂操作的工具如“先搜索再下载最后分析”。工具使用权限与安全为工具设置权限等级防止危险操作。3. 记忆系统的深化实现向量记忆Vector Memory使用向量数据库存储过去的对话和知识并能根据当前问题智能检索相关记忆实现“长期记忆”。总结性记忆Summary Memory定期对长对话进行总结将摘要存入长期记忆细节丢弃以节省Token。4. 可靠的错误处理与回退工具调用重试网络工具调用失败时自动重试若干次。备用工具当首选工具失败时自动尝试功能相似的备用工具。人工接管Human-in-the-loop在关键决策点或多次失败后暂停并请求人工干预。5. 可观测性与评估完整的日志系统记录每一步的输入、输出、工具调用、耗时和Token使用便于调试和成本分析。Agent表现评估设计自动化测试用例评估Agent完成特定任务的成功率、步骤数和结果质量。搭建Agent框架就像教一个聪明的孩子学会使用各种工具来完成项目。初期它可能会犯很多错比如用错工具、步骤混乱。但通过清晰的设计规划器、丰富的工具库、可靠的记忆系统和严谨的执行逻辑你可以逐步引导它变得可靠、高效。这个过程没有银弹需要你根据具体的业务场景反复迭代和调优。希望这个从零开始的指南能为你打开自主Agent世界的大门剩下的精彩就等你用代码去实现了。
)
