从零搭建AI Agent工具链：手写Python智能体实现ReAct与Function Calling

最近业务中要做一个内部提效工具：让大模型根据自然语言自动查数据库、读文件、调内部接口，最后产出一份分析报告。刚开始我直接用脚本拼 Prompt，发现完全跑不通，模型经常答非所问，工具调用也总是出格式错误。后来把整个链路拆成“模型调用 + 工具注册 + 记忆管理 + 编排策略”一套工具链，才逐渐稳定。

这篇文章就把我从零搭建智能体工具链的完整过程分享出来。不依赖 Dify、Coze 这类平台，而是用 Python 从模型接入层开始，手写一个轻量但可扩展的 Agent 框架，最终实现带 Function Calling、工具注册、记忆裁剪和多 Agent 编排的完整体系。

1. Agent 到底是什么：概念拆解与工具链全景

1.1 什么是 AI Agent

AI Agent（智能体）是当前大模型应用里最热的方向之一。通俗地解释，Agent 是一个“能自己决定下一步做什么”的 AI 程序。

普通的大模型调用是这样的：

用户输入 -> 模型 -> 文本输出

模型只能“说”，不能“做”。而 Agent 是：

用户输入 -> 模型规划 -> 调用工具 -> 观察结果 -> 继续规划 -> 最终输出

模型不仅能理解你的问题，还能主动决定要不要调用某个工具、调用哪个工具、传什么参数。调用完后，把结果继续喂给模型，模型再判断下一步是继续调用还是给最终答案。

这个循环就是 Agent 最核心的机制，业界通常叫它ReAct（Reasoning + Acting），也就是“推理 + 行动”交替进行。

1.2 为什么需要一套完整的工具链

只给模型一个工具函数，不叫 Agent。真正的 Agent 开发涉及以下几个环节：

• 模型接入层：不同厂商的模型接口、不同模型版本，需要统一封装。
• 工具系统：工具如何定义、注册、校验参数、执行、返回结果。
• 记忆模块：保持多轮对话上下文，必要时支持长期记忆。
• 编排策略：单个 Agent 能力有限，多个 Agent 如何分工协作。
• 安全边界：工具权限控制、敏感操作确认、输出内容过滤。

这些环节共同构成一套“智能体工具链”。没有这套工具链，开发 Agent 就是写一次性脚本；有了这套工具链，你可以在不同项目里复用同一套能力。

1.3 从零搭建 vs 使用现成平台

现在的选择其实挺多：Dify、Coze 都能快速搭建 Agent，也有一些开源的 Agent 框架。但如果你需要深度定制、离线部署、学习底层机制，从零搭建反而是更稳的选择：

• 可控性强：每一层逻辑都知道是干什么的，出了问题好排查。
• 依赖少：只依赖模型 API，不受平台规则限制。
• 学习价值高：能彻底理解 Agent 内部的消息流转机制。

本文从零搭建不是“重复造轮子”，而是先理解原理，再决定要不要引入更重的框架。

2. 整体架构设计：Agent 工具链的分层模型

2.1 五层架构模型

一套完整的 Agent 工具链，我习惯分成五层：

每层只负责自己的事情。例如工具层不关心模型是 GPT 还是 DeepSeek，只要模型能输出标准的工具调用格式就行；编排层也不关心工具内部怎么实现，只看工具返回的结果。

2.2 核心流程：感知—规划—行动—观察

Agent 的单次运行循环可以拆成这样：

第 1 步：接收用户输入 第 2 步：把当前对话历史 + 工具清单发给模型 第 3 步：模型决定直接回答，或者调用某个工具 第 4 步：执行工具，拿到结果 第 5 步：把结果追加到对话历史 第 6 步：回到第 2 步，直到模型给出最终结果或达到最大步数

注意第 5 步特别关键：每次工具调用结果都必须以tool角色追加到消息列表里，模型才能看到执行结果。

2.3 项目目录结构规划

本文实战项目叫min-agent，目录结构如下：

min-agent/ ├── agent.py # Agent 执行器（ReAct 循环） ├── llm.py # 模型接入层封装 ├── tools.py # 工具注册与定义 ├── memory.py # 记忆管理模块 ├── orchestration.py # 多 Agent 编排 ├── config.py # 配置读取 ├── main.py # 入口演示 └── requirements.txt # 依赖

后面的章节会按这个结构逐个文件实现。

3. 环境准备与依赖说明

3.1 技术选型

本文实现基于 Python 3.10+，模型接入使用 OpenAI 兼容接口。之所以选 OpenAI 兼容协议，是因为国内很多模型服务商（DeepSeek、通义千问、Moonshot 等）都提供该协议，代码可以无缝切换。

你需要准备：

• Python 3.10 或更高版本。
• 一个支持 Function Calling / Tools 的大模型 API Key。
• 能用 pip 安装依赖的网络环境。

3.2 安装依赖

新建虚拟环境后，安装下面几个依赖：

pip install openai python-dotenv

这里不引入任何 Agent 框架，只用最基础的 SDK。

3.3 配置管理

在项目根目录建.env文件：

LLM_BASE_URL=https://api.example.com/v1 LLM_API_KEY=sk-your-key LLM_MODEL=gpt-4o-mini

注意：LLM_BASE_URL请填写你实际使用的模型服务地址。如果用 DeepSeek，就填 DeepSeek 的接口地址；如果用通义，就填通义的兼容地址。具体参数以各家服务商文档为准。

config.py读取配置：

import os from dotenv import load_dotenv load_dotenv() class Config: LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://api.example.com/v1") LLM_API_KEY = os.getenv("LLM_API_KEY", "") LLM_MODEL = os.getenv("LLM_MODEL", "gpt-4o-mini") MAX_STEPS = int(os.getenv("MAX_STEPS", "5")) MAX_HISTORY = int(os.getenv("MAX_HISTORY", "20")) config = Config()

4. 模型接入层与消息协议

4.1 统一的消息格式

所有模型调用都围绕一个消息列表messages展开。消息有三种常见角色：

• system：系统提示词，定义 Agent 的行为。
• user：用户输入。
• assistant：模型回复。当模型发起工具调用时，这个回复里会包含tool_calls字段。
• tool：工具执行结果。

llm.py封装一个ChatClient，屏蔽不同模型的差异：

from openai import OpenAI from config import config class ChatClient: def __init__(self): self.client = OpenAI( base_url=config.LLM_BASE_URL, api_key=config.LLM_API_KEY, ) self.model = config.LLM_MODEL def chat(self, messages, tools=None): kwargs = { "model": self.model, "messages": messages, } if tools: kwargs["tools"] = tools response = self.client.chat.completions.create(**kwargs) return response.choices[0].message

这段代码最核心的是：当传入tools时，模型具备了“决定是否调用工具”的能力。工具清单会在 API 请求里以 JSON Schema 的形式传给模型。

4.2 系统提示词的设计

系统提示词决定了 Agent 的“人设”和“行为边界”。下面是一个基础模板：

SYSTEM_PROMPT = """ 你是一个智能助手，你可以通过调用工具来完成用户的任务。 使用规则： 1. 当用户的问题需要实时数据或执行操作时，使用工具。 2. 当工具返回结果后，基于结果回答用户。 3. 如果不需要工具，直接回答。 4. 所有回答使用中文。 """

这个提示词要放在messages列表的第一条。

5. 工具系统：ToolRegistry 的设计与实现

5.1 为什么需要工具注册中心

一个 Agent 会挂载很多工具。如果直接在代码里写 if-else 判断工具名称，工具一多就会乱。工具注册中心解决几个问题：

• 统一管理工具清单。
• 自动生成模型要求的 JSON Schema。
• 按名称快速找到并执行工具。

5.2 核心代码实现

tools.py完整实现：

import json class Tool: def __init__(self, name, description, schema, func): self.name = name self.description = description self.schema = schema # JSON Schema self.func = func def run(self, **kwargs): return self.func(**kwargs) class ToolRegistry: def __init__(self): self._tools = {} def register(self, tool: Tool): self._tools[tool.name] = tool def unregister(self, name: str): self._tools.pop(name, None) def get(self, name: str): tool = self._tools.get(name) if not tool: raise KeyError(f"工具未注册: {name}") return tool def schemas(self): schemas = [] for tool in self._tools.values(): schemas.append({ "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.schema, }, }) return schemas registry = ToolRegistry()

5.3 内置工具示例

实现两个工具：获取当前时间、写文件。

import datetime def get_current_time(): """获取当前时间""" now = datetime.datetime.now() return {"time": now.strftime("%Y-%m-%d %H:%M:%S")} def save_to_file(path, content): """写入内容到指定文件""" with open(path, "w", encoding="utf-8") as f: f.write(content) return {"status": "ok", "path": path, "length": len(content)} registry.register(Tool( name="get_current_time", description="获取当前的系统时间，当用户询问日期或时间时调用", schema={"type": "object", "properties": {}}, func=get_current_time, )) registry.register(Tool( name="save_to_file", description="将文本内容保存到本地文件，当用户需要记录备忘或生成文件时调用", schema={ "type": "object", "properties": { "path": {"type": "string", "description": "保存路径"}, "content": {"type": "string", "description": "文件内容"}, }, "required": ["path", "content"], }, func=save_to_file, ))

注意工具的参数 Schema 必须写严格。模型会根据这个 Schema 生成参数，如果字段缺失或类型错误，工具执行时就会报错。

5.4 工具系统的扩展思路

实际项目中，你可以在Tool.run()里加入参数校验、超时控制、权限检查，也可以在工具返回时统一包一层结构：

def safe_run(self, **kwargs): try: result = self.func(**kwargs) return {"success": True, "result": result} except Exception as e: return {"success": False, "error": str(e)}

这样工具内部即使报错，也不会把 Agent 整个流程打挂。

6. Agent 执行器：实现 ReAct 循环

6.1 执行器的完整代码

agent.py是整套工具链的心脏，实现前面讲的“推理—行动—观察”循环：

import json from llm import ChatClient from tools import registry from memory import Memory from config import config class Agent: def __init__(self, system_prompt, name="agent"): self.name = name self.client = ChatClient() self.system_prompt = system_prompt self.max_steps = config.MAX_STEPS self.memory = Memory(max_messages=config.MAX_HISTORY) self.tools = registry def run(self, user_input): self.memory.add_user(user_input) messages = [{"role": "system", "content": self.system_prompt}] messages.extend(self.memory.history) for step in range(1, self.max_steps + 1): print(f"[{self.name}] 第 {step} 步: 调用模型") message = self.client.chat(messages, tools=self.tools.schemas()) if message.tool_calls is None: # 模型没有要求调用工具，直接作为最终回答 self.memory.add_assistant(message.content) return message.content # 模型要求调用工具 messages.append(message) for tool_call in message.tool_calls: fn_name = tool_call.function.name fn_args = json.loads(tool_call.function.arguments) print(f"[{self.name}] 调用工具: {fn_name}, 参数: {fn_args}") tool = self.tools.get(fn_name) result = tool.run(**fn_args) # 工具结果追加回消息列表，模型在下一次迭代中可以看到 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False), }) self.memory.add_tool_result(fn_name, result) return "已达到最大执行步数，任务可能未完成。"

这个循环的关键在于：

• 每轮模型返回结果有两种可能：要么是普通文字，要么是tool_calls。
• 只要模型返回tool_calls，就执行工具，把结果塞回messages，然后继续下一轮。
• 当模型不再要求调用工具时，说明它已经拿到足够信息，可以输出最终答案了。

6.2 每一步都在做什么

以“现在几点了？帮我记录到笔记.txt”为例：

第 1 轮：模型看到工具列表，判断“现在几点”需要 get_current_time，返回工具调用。 执行 get_current_time，拿到时间。 第 2 轮：模型看到时间结果，判断“记录到笔记.txt”需要 save_to_file。 执行 save_to_file，拿到保存结果。 第 3 轮：模型看到保存结果，不再要求工具，返回最终回答。

这就是 ReAct 循环的直观理解。

7. 记忆模块：短期记忆与上下文管理

7.1 为什么 Agent 需要记忆

Agent 是多轮对话系统。如果没有记忆，模型每轮只能看到当前输入，之前聊过的内容全部丢失。比如用户先说“帮我查下北京的天气”，再说“那上海呢？”，模型必须记得上一轮聊的是天气。

memory.py实现一个简单的短期记忆：

class Memory: def __init__(self, max_messages=20): self.history = [] self.max_messages = max_messages def add_user(self, content): self.history.append({"role": "user", "content": content}) self._trim() def add_assistant(self, content): self.history.append({"role": "assistant", "content": content}) self._trim() def add_tool_result(self, tool_name, result): text = f"工具 {tool_name} 返回：{result}" self.history.append({"role": "system", "content": text}) self._trim() def clear(self): self.history = [] def _trim(self): if len(self.history) > self.max_messages: self.history = self.history[-self.max_messages:]

注意_trim()每次追加消息后都会检查长度。对于上下文窗口有限的模型，这个裁剪策略很关键。

7.2 记忆裁剪策略的选择

裁剪策略有很多种：

• 保留最近 N 条：最简单，但可能丢失早期关键信息。
• 摘要历史：每轮对话后让模型生成摘要，保留摘要 + 最近对话。
• 向量检索：把历史存入向量库，按相关性召回。

实际项目里，摘要和向量检索更实用。本文先实现前两种策略中的“最近 N 条”，后续可以平滑替换为向量检索。

7.3 长期记忆的简要思路

如果要做长期记忆，可以引入一个简单的 JSON 文件作为存储。每次 Agent 启动时加载，对话结束后写入。比如：

import json import os class JsonMemory: def __init__(self, store_path="memory_store.json"): self.store_path = store_path self.data = {} self._load() def _load(self): if os.path.exists(self.store_path): with open(self.store_path, "r", encoding="utf-8") as f: self.data = json.load(f) def save(self): with open(self.store_path, "w", encoding="utf-8") as f: json.dump(self.data, f, ensure_ascii=False, indent=2) def put(self, key, value): self.data[key] = value self.save() def get(self, key, default=None): return self.data.get(key, default)

这个模块适合存储用户偏好、历史任务记录等结构化信息。

8. 多 Agent 编排：让多个智能体协同工作

8.1 为什么需要多 Agent

单一 Agent 适合简单任务，但复杂问题最好拆解。例如“写一篇技术博客”这个任务，可以拆成“研究主题”和“撰写文章”两个角色。多 Agent 编排就是让多个各司其职的 Agent 配合。

8.2 简单编排实现

orchestration.py：

from agent import Agent from tools import registry class Orchestrator: def __init__(self, agents: dict): self.agents = agents def route(self, user_input: str) -> str: # 简单的路由规则：根据关键词选择 Agent if any(kw in user_input for kw in ["写", "生成", "总结"]): return "writer" if any(kw in user_input for kw in ["时间", "几点", "日期"]): return "time" return "general" def run(self, user_input: str): agent_name = self.route(user_input) print(f"[orchestrator] 路由到 {agent_name}") agent = self.agents[agent_name] return agent.run(user_input)

这段代码是演示性质，真实场景中可以用模型做路由判断，也可以用更复杂的规则引擎。重点是编排层只负责“派任务”，不关心 Agent 内部实现。

8.3 多个 Agent 的执行流程

假设创建了time_agent、writer_agent、general_agent三个 Agent，分别挂载不同工具。用户输入“现在几点，顺便帮我生成一个今天的工作总结”，编排器先路由，再决定是顺序调用还是并行调用。实际项目可以根据任务依赖关系灵活设计。

9. 完整运行与验证

9.1 入口文件

main.py：

from agent import Agent from orchestration import Orchestrator from tools import registry import tools # noqa: F401 确保工具注册执行 def create_agents(): time_agent = Agent( system_prompt="你是一个时间助手，负责查询时间。", name="time_agent", ) general_agent = Agent( system_prompt="你是一个通用助手，基于工具结果回答用户问题。", name="general_agent", ) return {"time": time_agent, "general": general_agent} if __name__ == "__main__": agents = create_agents() orch = Orchestrator(agents) while True: user_input = input("你: ") if user_input.strip().lower() in {"exit", "quit", "q"}: break response = orch.run(user_input) print(f"Agent: {response}")

运行：

python main.py

预期交互：

你: 现在几点？ [orchestrator] 路由到 time [time_agent] 第 1 步: 调用模型 [time_agent] 调用工具: get_current_time, 参数: {} [time_agent] 第 2 步: 调用模型 Agent: 当前时间是 2025-06-01 14:23:05。

9.2 验证工具调用逻辑

如果你想不依赖真实模型，也可以用一段模拟函数来验证流程。用 mock 数据测试时，重点观察消息列表中role是否交替正确。这是 Agent 开发中最容易出错的地方。

10. 常见问题与排查思路

排查 Agent 问题有一个万能套路：把传给模型的messages原样打印出来，看每一轮的消息是否完整、顺序是否正确。90% 的问题都出在消息列表上。

11. 最佳实践与工程建议

11.1 工具设计原则

• 每个工具只做一件事，不要写“万能工具”。
• 工具描述要具体，说明什么情况下使用，模型才能正确选择。
• 参数 Schema 的description写清楚，模型才能生成正确参数。
• 工具必须有超时和异常处理，不能因为一个工具失败拖垮整个 Agent。

11.2 提示词工程建议

• 系统提示词里明确“工具优先还是直接回答”。
• 控制模型的自由发挥空间，用“必须基于工具结果回答”这类约束。
• 不要把所有规则都堆进提示词，能写在代码里的逻辑不要写进提示词。

11.3 生产环境安全边界

• 权限最小化：Agent 挂载的文件写入工具，应该限制可写目录，而不是任意路径。
• 敏感操作确认：删除文件、发送消息、执行命令等工具，需要二次确认。
• 日志记录：每个工具调用都要记录调用方、参数、结果、耗时，便于审计。
• 速率限制：Agent 循环会频繁调用模型，要考虑 API 限流和成本控制。

11.4 可观测性

开发 Agent 最容易遇到“黑盒问题”。建议在关键节点打日志：

[time_agent] 第 1 步: 调用模型 [time_agent] 调用工具: get_current_time, 参数: {} [time_agent] 工具返回: {"time": "2025-06-01 14:23:05"}

这套日志能直观看到 Agent 每一步在做什么，排查问题效率会高很多。

12. 总结与学习路线

这套工具链从模型接入层、工具注册中心、ReAct 执行器、记忆模块到多 Agent 编排，已经覆盖了 Agent 开发的核心链路。建议按下面的路线继续深入：

• 先把本文代码完整跑通，理解messages的流转过程。
• 给 ToolRegistry 增加参数校验和权限控制。
• 引入向量数据库实现长期记忆。
• 尝试把编排策略从规则升级为模型判断。
• 扩展 Agent 的工具，比如 HTTP 请求、数据库查询、代码执行。

下一步想深入的话，重点研究两个方向：一是 Function Calling 在不同模型上的差异，二是多 Agent 协作中的任务拆分与结果合并。这两个方向是 Agent 开发走向工程化的必经之路。