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）。

为什么需要规划器？直接让大模型输出最终答案，对于复杂任务来说失败率极高。规划器迫使模型进行“逐步推理”。例如，目标“写一份关于新能源汽车的市场分析报告”。一个简单的规划器可能会引导模型产出如下步骤：

1. 搜索并收集近期新能源汽车行业的政策、销量数据、头部公司动态。
2. 对收集到的信息进行归纳整理，提炼出核心趋势、挑战与机遇。
3. 根据分析结果，搭建报告框架（引言、市场现状、竞争分析、未来展望等）。
4. 撰写报告正文。
5. 检查并润色报告。

如何实现？最简单的实现方式是设计一个特定的“系统提示词”（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 执行引擎与调度器：让一切运转起来的“心脏”

这是框架的驱动模块。它负责串联整个流程：

1. 接收目标：获取用户输入。
2. 初始化规划：调用规划器，生成任务列表，初始化状态。
3. 循环执行： a.任务选择：根据状态，选择下一个待执行的任务。 b.推理与工具调用：将当前任务描述、历史上下文、可用工具列表传给大模型。模型决定是进行“思考”（输出文本）还是“行动”（调用工具）。 c.执行工具：如果模型决定调用工具，调度器则找到对应工具函数，传入参数并执行。 d.结果处理：将工具执行结果或模型思考内容，更新到状态和记忆中。 e.状态检查：判断当前任务是否完成，是否所有任务已完成，或是否出现错误需要调整计划。
4. 输出最终结果：所有任务完成后，整合信息，生成最终输出给用户。

调度器还需要处理错误和异常，比如工具调用失败、模型返回格式错误等，并设计重试或回退策略。

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_results=max_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代码。必须替换为安全的方案，例如：

1. 使用ast.literal_eval()处理纯数字和运算符。
2. 使用专门的数学表达式解析库（如numexpr）。
3. 在严格的沙箱环境（如restrictedpython）中执行。在你的项目中，请务必采用安全方案。

第二步：定义记忆与状态

from dataclasses import dataclass, field from typing import List, Dict, Any @dataclass class AgentState: """Agent执行状态""" goal: str # 最终目标 plan: List[Dict] = field(default_factory=list) # 任务计划，每个任务是一个dict completed_tasks: List[Dict] = field(default_factory=list) # 已完成的任务及结果 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 None

3.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_key=api_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, tools=tools_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(goal=goal, plan=tasks) # 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_key=api_key, model="gpt-3.5-turbo", tools=tools) # 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：陷入死循环或步骤混乱。

• 可能原因：状态管理逻辑有缺陷。例如，任务完成条件判断不准。解决：
  1. 引入更明确的任务完成标志。可以让模型在输出中明确声明“任务完成”。
  2. 设置最大步数（max_steps）作为安全阀。
  3. 实现“反思”步骤：每执行几步后，让模型回顾一下进度和计划是否需要调整。

问题4：Token消耗巨大，成本高。

• 可能原因：上下文（context）无限增长。解决：
  1. 压缩历史：不是存储完整的对话记录，而是定期让模型总结之前的交互，用总结代替详细历史。
  2. 选择性记忆：只保留与当前任务最相关的历史片段。可以利用向量数据库检索相关记忆。
  3. 使用更经济的模型：规划、工具选择等步骤可以使用小模型（如GPT-3.5），最终汇总答案再用大模型（如GPT-4）。

4.2 从Demo到生产：关键优化点

上面的框架是一个教学原型。要用于实际项目，你需要考虑以下方面：

1. 更强大的规划与反思（ReAct, Plan-and-Execute）

• 实现ReAct（Reasoning + 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世界的大门，剩下的精彩，就等你用代码去实现了。