这次我们来看一个偏“硬核”的方向不用任何现成 Agent 框架从零搭建一套属于自己的智能体工具链。所谓 Agent本质上就是让大模型不只停留在“对话”而是具备工具调用、任务拆解、记忆管理和自主执行的能力。市面上的 LangChain、AutoGPT、MetaGPT 等框架很多但如果只是调用框架 API很多底层机制其实是黑盒。自己动手搭一遍之后再回头看这些框架的设计思路会通透很多。这篇文章会从工具链的整体架构讲起然后直接进入代码层面实现一个最小可运行的 Agent 核心LLM 接口接入、工具注册与调用、短期/长期记忆、执行循环、批量任务调度。还会给出 API 接入示例、资源占用观察方法、常见问题排查清单和工程化最佳实践。适合以下读者想理解 Agent 工作原理的开发者准备做 Agent 相关面试或项目沉淀的工程师需要把 Agent 接入业务系统做自动化测试、内容生成、批量任务处理的技术人员。不需要有框架使用经验但要会 Python 基础并且至少有一个可调用的 LLM API 密钥。1. 核心能力速览能力项说明项目类型智能体Agent开发工具链从零手写核心机制核心能力工具注册与调用、任务拆解、多轮执行循环、记忆管理依赖模型支持 OpenAI 兼容接口的 LLM如 DeepSeek、GPT 系列等开发语言Python 3.10启动方式命令行启动可封装为 API 服务是否支持 CPU取决于所选 LLM 是云端 API 还是本地模型显存需求若调用云端 API 则本机基本无显存压力若本地部署 7B/14B 模型需按模型参数量预留显存是否支持批量任务支持可通过并发队列批量执行是否支持 API支持可将 Agent 封装为 HTTP 服务适合场景自动化测试、内容生成、信息检索、数据分析、内部工具串联2. 适用场景与使用边界自己搭建 Agent 工具链意味着你不受某个框架绑定可以在核心循环上做任意定制。比较典型的使用场景包括自动化测试Agent 根据需求描述自动生成测试用例、调用测试工具、分析失败日志。内容生成管线Agent 调用搜索工具、资料库、写作模型完成从资料收集到成稿的全流程。数据报表分析Agent 连接数据库工具、代码执行工具按用户提问生成分析结果。内部工具串联把公司内部 API 注册成工具让 Agent 按权限调用。但 Agent 不是万能的。它不适合以下场景对实时性要求极高的系统模型推理耗时叠加工具调用耗时可能达到数秒甚至数十秒。完全离线、不允许任何外部 API 调用的内网环境如果用云端大模型数据流转到第三方需要严格评估。高风险决策场景医疗诊断、金融交易等Agent 的幻觉风险不可接受。合规边界必须提前确认所有工具调用要遵循最小权限原则Agent 只能访问它完成任务所必需的资源。涉及用户隐私数据、业务数据的场景必须先获得合法授权且数据处理链路要符合相关法规。不能把 Agent 用于绕过安全限制、攻击系统、窃取账号或批量爬取他人数据。如果 Agent 生成的内容对外发布或商用必须做人工复核。3. 环境准备与前置条件3.1 开发环境检查清单在开始写代码之前先确认以下环境是否就绪项目建议要求说明操作系统Windows 10/11、macOS、Linux核心代码跨平台Python3.10 或更高建议使用 3.11性能和兼容性均衡依赖管理venv 或 conda建议每个项目隔离虚拟环境LLM APIOpenAI 兼容接口的 API KeyDeepSeek、GPT 等均可网络可正常访问 API 服务如果部署本地模型则不受此限制磁盘空间至少 5GB 空闲含依赖和日志空间端口避免 8000、8080、7860 被占用如做 API 服务需提前确认3.2 创建项目目录mkdir agent-toolchain cd agent-toolchain python -m venv venv source venv/bin/activate # Windows 下则执行 venv\Scripts\activate3.3 安装基础依赖pip install openai python-dotenv pydantic requests rich说明openai官方 Python SDK 可以调用 OpenAI 兼容接口DeepSeek 等国产模型也支持同一协议。python-dotenv用于管理环境变量避免把 API Key 写死在代码里。pydantic做数据校验和工具参数 schema 校验。rich用于终端日志输出美化方便观察 Agent 执行过程。4. Agent 工具链的总体架构手写 Agent 工具链最少需要六个模块用户输入 → 规划器 → 工具调度 → 执行器 → 反馈处理器 → 输出结果 ↑______ 记忆模块 ←______|4.1 六个核心模块规划器Planner接收用户目标由 LLM 决定下一步行动。可能拆解为多步任务也可能直接选择工具。工具注册表Tool RegistryAgent 可调用的所有工具的元信息集合。每个工具包含名称、描述、参数 schema、执行函数。LLM 根据描述决定调哪个工具。执行器Executor实际执行工具函数的代码逻辑。注意工具函数在本地运行LLM 只负责给出工具名和参数不直接执行。反馈处理器Feedback Handler把工具执行结果返回给 LLM让模型判断下一步动作。记忆模块Memory Module短期记忆保留当前任务上下文长期记忆按需存储历史事实和偏好。输出解析器Output Parser把 LLM 的响应解析为结构化指令。这是最容易出错的一环也是最需要自定义的地方。4.2 一次完整执行流程第 1 轮用户输入查询北京今天的天气 第 2 轮规划器决定调用 weather.get_current_weather 工具参数 city北京 第 3 轮执行器本地执行天气查询函数返回北京今天晴22 度 第 4 轮反馈处理器将天气结果返回给 LLM 第 5 轮LLM 根据结果生成最终回答这套循环就是 Agent 的“ReAct 模式”Reasoning推理 Acting行动。整个 Agent 引擎的核心就是把这个循环跑稳。5. 从零实现 Agent 核心代码5.1 定义工具基类工具是整个 Agent 的能力来源。设计成装饰器注册模式最灵活# tools.py import inspect from typing import Any, Callable, Dict, List class Tool: def __init__(self, name: str, description: str, func: Callable, parameters: Dict[str, Any]): self.name name self.description description self.func func self.parameters parameters def run(self, **kwargs): return self.func(**kwargs) def to_openai_schema(self) - Dict[str, Any]: return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, } } _TOOL_REGISTRY: Dict[str, Tool] {} def register_tool(name: str, description: str, parameters: Dict[str, Any]): def decorator(func): _TOOL_REGISTRY[name] Tool( namename, descriptiondescription, funcfunc, parametersparameters, ) print(f[Tool Registry] 已注册工具: {name}) return func return decorator def get_tool(name: str) - Tool: if name not in _TOOL_REGISTRY: raise KeyError(f工具不存在: {name}) return _TOOL_REGISTRY[name] def list_tools() - List[Tool]: return list(_TOOL_REGISTRY.values())这里的关键设计点每个工具都有标准化的to_openai_schema()方法可以直接转换成 OpenAI 函数调用协议需要的 JSON Schema。5.2 定义示例工具下面注册两个最常用工具加法和查询时间。# main.py import datetime from tools import register_tool, list_tools register_tool( namecalculator.add, description计算两数相加, parameters{ type: object, properties: { x: {type: number, description: 第一个数字}, y: {type: number, description: 第二个数字} }, required: [x, y] } ) def add(a: float, b: float) - float: return a b register_tool( namesystem.current_time, description获取当前系统时间, parameters{ type: object, properties: {} } ) def current_time() - str: return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) if __name__ __main__: tools list_tools() print(f当前注册了 {len(tools)} 个工具)运行一次确认工具注册正常这是后续所有逻辑的地基。5.3 实现 LLM 客户端封装封装一个ChatClient使用 openai SDK 调用 OpenAI 兼容接口。# llm_client.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() class ChatClient: def __init__(self, model: str deepseek-chat): self.client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.deepseek.com/v1), ) self.model model def chat(self, messages: list, tools: list None) - dict: kwargs {model: self.model, messages: messages} if tools: kwargs[tools] tools kwargs[tool_choice] auto response self.client.chat.completions.create(**kwargs) message response.choices[0].message return { content: message.content, tool_calls: message.tool_calls, }这里有几个细节base_url配置成环境变量方便切换不同模型服务商。tool_choiceauto让模型自主决定调用哪个工具。返回的tool_calls就是模型给出的“工具调用指令”。5.4 实现 Agent 执行循环这是整套工具链最核心的部分。执行循环要做的事发送消息给 LLM。判断是否有工具调用。如果有执行工具将结果追加到消息记录返回 LLM。如果没有工具调用将 content 作为最终回答返回。# agent.py import json from llm_client import ChatClient from tools import get_tool, list_tools class Agent: def __init__(self, system_prompt: str, max_iters: int 10): self.client ChatClient() self.system_prompt system_prompt self.max_iters max_iters self.messages [ {role: system, content: self.system_prompt} ] def run(self, user_input: str) - str: self.messages.append({role: user, content: user_input}) tool_schemas [tool.to_openai_schema() for tool in list_tools()] for step in range(self.max_iters): print(f\n Step {step 1} ) result self.client.chat(self.messages, toolstool_schemas) # 情况一模型没有发起工具调用直接返回结果 if not result[tool_calls]: answer result[content] or 模型未返回内容 self.messages.append({role: assistant, content: answer}) return answer # 情况二模型发起工具调用 assistant_msg {role: assistant, content: None} tool_calls [] for tc in result[tool_calls]: tool_calls.append({ id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, } }) print(f[Agent] 调用工具: {tc.function.name} 参数: {tc.function.arguments}) assistant_msg[tool_calls] tool_calls self.messages.append(assistant_msg) # 逐个执行工具并把结果写回消息记录 for tc in tool_calls: func_name tc[function][name] args_str tc[function][arguments] try: args json.loads(args_str) tool_func get_tool(func_name) output tool_func.run(**args) output_str json.dumps(output, ensure_asciiFalse) except Exception as e: output_str f工具执行出错: {str(e)} print(f[Tool] {func_name} 返回: {output_str[:100]}) self.messages.append({ role: tool, tool_call_id: tc[id], content: output_str, }) raise TimeoutError(f超过最大执行轮数 {self.max_iters} 次任务中止)这个循环已经能跑通最基础的 Agent 能力。注意几个容易踩的坑工具调用后必须追加role: tool的消息并且tool_call_id必须对上。执行工具前要做 try-except防止单个工具报错直接让整个 Agent 崩溃。max_iters必须限制否则模型可能陷入无限循环。5.5 接入记忆模块记忆由两种短期记忆就是上面的self.messages上下文窗口长期记忆需要独立存储。简单实现一个基于 JSON 文件的长期记忆# memory.py import json import os from datetime import datetime class MemoryStore: def __init__(self, memory_file: str memory.json): self.memory_file memory_file self._ensure_file() def _ensure_file(self): if not os.path.exists(self.memory_file): with open(self.memory_file, w, encodingutf-8) as f: json.dump({}, f, ensure_asciiFalse, indent2) def save(self, key: str, value: str): data self._load() data[key] { value: value, updated_at: datetime.now().isoformat() } self._write(data) def load(self, key: str) - str or None: data self._load() item data.get(key) return item[value] if item else None def _load(self): with open(self.memory_file, r, encodingutf-8) as f: return json.load(f) def _write(self, data): with open(self.memory_file, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2)在 Agent 类中把记忆模块组合进去每次对话开始时检查是否有历史的偏好或事实可以注入 system prompt。这里不做太复杂的向量检索先保证“能存能取”。6. Agent 接口 API 与批量任务6.1 将 Agent 封装为 FastAPI 服务工具链搭好之后如果要接业务系统需要把 Agent 暴露成 HTTP API。用 FastAPI 是最快的方式pip install fastapi uvicorn# api_service.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent import Agent app FastAPI(titleAgent Toolchain API, version0.1.0) class AgentRequest(BaseModel): input: str session_id: str default class AgentResponse(BaseModel): output: str session_prompts {} app.post(/agent/run, response_modelAgentResponse) def run_agent(req: AgentRequest): try: if req.session_id not in session_prompts: session_prompts[req.session_id] ( 你是智能助手任务复杂时请拆解为多个步骤使用工具完成。 ) agent Agent(system_promptsession_prompts[req.session_id]) output agent.run(req.input) return AgentResponse(outputoutput) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/agent/tools) def list_agent_tools(): return {tools: [t.to_openai_schema() for t in list_tools()]} if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动命令uvicorn api_service:app --host 127.0.0.1 --port 8000启动后可以直接访问http://127.0.0.1:8000/docs查看 Swagger 接口文档。接口可以跑通后面就可以把 Agent 能力接到自己的业务工具里了。6.2 curl 调用示例curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {input: 请计算 123 456然后告诉我结果, session_id: test-001}预期返回{ output: 123 456 的结果是 579。 }6.3 Python 批量任务调度Agent 服务化之后批量任务就是并发调用的问题。写一个简单的批量调度器# batch_runner.py import json import requests from concurrent.futures import ThreadPoolExecutor, as_completed API_URL http://127.0.0.1:8000/agent/run def run_one(input_text: str, session_id: str) - dict: payload {input: input_text, session_id: session_id} try: resp requests.post(API_URL, jsonpayload, timeout120) resp.raise_for_status() return {ok: True, input: input_text, data: resp.json()} except Exception as e: return {ok: False, input: input_text, error: str(e)} def batch_run(tasks: list[dict], max_workers: int 4): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [ executor.submit(run_one, task[input], task.get(session_id, batch)) for task in tasks ] for future in as_completed(futures): results.append(future.result()) return results if __name__ __main__: tasks [ {input: 计算 1 1, session_id: batch-1}, {input: 计算 10 20, session_id: batch-2}, {input: 计算 100 200, session_id: batch-3}, ] results batch_run(tasks, max_workers3) for r in results: print(json.dumps(r, ensure_asciiFalse, indent2))批量任务的几个要点max_workers不建议开太大因为 LLM API 通常有并发限制且大批量并发会导致延时不稳。每个任务要有独立 session_id避免上下文污染。单个任务必须设置超时时间否则某个任务卡住会占用线程池。批量任务建议先跑 3 到 5 条测试确认接口稳定后再全量跑。7. 资源占用与性能观察7.1 调用云端 API 的资源特点如果你用的是 DeepSeek、GPT 等云端 API本机资源占用非常小主要是 Python 进程内存和网络带宽。一个并发数为 4 的批量调度进程内存通常在 200MB 到 500MB 之间。这种方式的好处是显存无压力劣势是每次调用都要付费、有网络延迟并且存在数据出域的合规风险。7.2 本地部署 LLM 的资源特点如果要完全本地部署需要一个支持 OpenAI 兼容接口的推理服务资源占用取决于模型参数量。一个 7B 参数的量化模型通常需要 6GB 到 10GB 显存14B 模型需要 15GB 以上70B 模型则需要多张显卡或者大内存配合 CPU 推理。实际显存占用必须以你选择的模型和推理框架为准不同量化级别差异很大。7.3 如何观察资源占用开发调试时可以用以下命令观察资源# Linux/macOS top -p pid # Python 脚本内查看内存 import psutil process psutil.Process() print(process.memory_info().rss / 1024 / 1024, MB)7.4 降低资源占用的策略Agent 循环中消息会不断累积建议对历史消息做截断只保留最近 N 轮。工具返回内容如果很大先做截断再塞回上下文。并发批量任务时控制max_workers避免同时创建大量 HTTP 连接。本地部署推理时启用 KV Cache 量化、Flash Attention 等特性但这需要推理框架支持。8. 常见问题与排查方法问题现象可能原因排查方式解决方案调用 API 返回 401API Key 错误或未加载打印环境变量确认.env文件存在重新配置LLM_API_KEYAPI 返回 429触发限流查看 API 返回错误码降低并发数加入重试退避机制Agent 不调用任何工具工具 schema 不合法或模型不认为需要工具打印 tool_schemas 检查格式修正 JSON Schema调低要求显式在 prompt 中提示模型使用工具工具执行报错参数类型不匹配打印 tool_calls 原始 arguments在工具函数内部做类型转换和默认值兜底tool_call_id不匹配assistant 消息和 tool 消息未正确对应检查消息记录顺序严格按“assistant 声明 tool_calls”后逐个追加 tool 响应Agent 陷入无限循环上下文太长或工具返回内容让模型误解查看日志判断重复行为设置 max_iters 上限让工具返回内容更清晰批量任务部分失败网络抖动或 API 限流查看失败任务的错误字段加入超时、重试和失败记录内存持续增长工具执行中对象未释放或消息列表过长监控 RSS 内存变化限制单次任务的消息数及时清理大对象端口被占用8000 被其他服务占用lsof -i :8000或netstat -ano换端口启动8.1 一个典型的 API 重试封装对于 429 和网络错误推荐在调用层加指数退避重试import time import random def request_with_retry(func, max_retries: int 3, base_delay: float 2.0): for attempt in range(max_retries): try: return func() except Exception as e: if attempt max_retries - 1: raise e delay base_delay * (2 ** attempt) random.uniform(0, 1) print(f请求失败{delay:.1f}s 后重试第 {attempt 1} 次: {e}) time.sleep(delay)9. 最佳实践与使用建议9.1 先小后大先慢后快第一次跑通 Agent 时先把 max_iters 设为 3工具只注册一个。等核心循环稳定了再逐步加工具、加并发、加记忆。9.2 保持最小可运行配置建议把“单工具调用”的用例固化为一个自动化测试。每次改动核心循环后先跑这个用例确认没有回归。9.3 工具函数做防御式编程Agent 传入的参数来自 LLM 生成非常不可控。工具函数内部不要信任输入尽量加类型校验、长度限制、空值兜底。比如日期工具要处理非法日期字符串文件工具要限制路径不能越界。9.4 日志和可观测性Agent 的调试难度比普通程序高因为中间多了一层模型推理不确定性。日志至少包含每次 LLM 调用的耗时和 token 数。工具调用名称、参数、返回内容摘要。消息上下文长度变化。错误堆栈和重试记录。建议以结构化 JSON 落盘方便事后排查。9.5 安全边界服务化后不要把 Agent 暴露到公网。如果必须暴露一定要加鉴权。工具的权限限制在最小范围内。比如 Agent 能调用数据库查询工具就只能用只读账号而不是管理员账号。凡是涉及程序执行、文件读写、命令行的工具必须做非常严格的输入校验和沙箱限制。涉及人脸、声音、版权素材、第三方数据的内容需要确认授权后方可使用和分发。9.6 成本控制云端 API 模式下Agent 每轮任务可能调用多次 LLM。一个多步任务可能消耗几千到几万 token。批量任务上线前建议先用一条样本估算 token 消耗。给每个会话设置 token 预算。对过长的工具返回内容做截断。保留每次调用的 token 使用记录按天汇总。10. 总结与下一步这套从零搭建的 Agent 工具链核心价值在于学会了“执行循环”的底层机制。真正把message - tool_calls - tool result - message这条闭环跑通之后你再去使用 LangChain、LangGraph 这类框架会清楚每个抽象层在做的事。先别急着加复杂功能。把最常见的问题验证完单工具调用、多工具连续调用、算数计算、时间查询跑通这四条就说明核心引擎是可用的。最容易踩的坑集中在工具参数解析和消息序列管理上这两个地方多花时间打磨。后面值得继续扩展的方向加入基于向量数据库的长期记忆。加入多 Agent 协作机制比如规划 Agent 和执行 Agent 分离。加入人工确认环节当工具调用风险较高时先暂停等用户确认再执行。加入评测集自动回归测试 Agent 在固定问题集上的稳定性。Agent 开发的核心不在于堆叠多少模型而在于把工具链的每个环节设计得可控、可观测、可恢复。从最小实现开始迭代是能找到瓶颈最快的方式。建议收藏备用动手写第一版循环的时候照着这份架构对照着做会比自己摸索快很多。
从零手写Agent工具链：核心机制与Python实践
这次我们来看一个偏“硬核”的方向不用任何现成 Agent 框架从零搭建一套属于自己的智能体工具链。所谓 Agent本质上就是让大模型不只停留在“对话”而是具备工具调用、任务拆解、记忆管理和自主执行的能力。市面上的 LangChain、AutoGPT、MetaGPT 等框架很多但如果只是调用框架 API很多底层机制其实是黑盒。自己动手搭一遍之后再回头看这些框架的设计思路会通透很多。这篇文章会从工具链的整体架构讲起然后直接进入代码层面实现一个最小可运行的 Agent 核心LLM 接口接入、工具注册与调用、短期/长期记忆、执行循环、批量任务调度。还会给出 API 接入示例、资源占用观察方法、常见问题排查清单和工程化最佳实践。适合以下读者想理解 Agent 工作原理的开发者准备做 Agent 相关面试或项目沉淀的工程师需要把 Agent 接入业务系统做自动化测试、内容生成、批量任务处理的技术人员。不需要有框架使用经验但要会 Python 基础并且至少有一个可调用的 LLM API 密钥。1. 核心能力速览能力项说明项目类型智能体Agent开发工具链从零手写核心机制核心能力工具注册与调用、任务拆解、多轮执行循环、记忆管理依赖模型支持 OpenAI 兼容接口的 LLM如 DeepSeek、GPT 系列等开发语言Python 3.10启动方式命令行启动可封装为 API 服务是否支持 CPU取决于所选 LLM 是云端 API 还是本地模型显存需求若调用云端 API 则本机基本无显存压力若本地部署 7B/14B 模型需按模型参数量预留显存是否支持批量任务支持可通过并发队列批量执行是否支持 API支持可将 Agent 封装为 HTTP 服务适合场景自动化测试、内容生成、信息检索、数据分析、内部工具串联2. 适用场景与使用边界自己搭建 Agent 工具链意味着你不受某个框架绑定可以在核心循环上做任意定制。比较典型的使用场景包括自动化测试Agent 根据需求描述自动生成测试用例、调用测试工具、分析失败日志。内容生成管线Agent 调用搜索工具、资料库、写作模型完成从资料收集到成稿的全流程。数据报表分析Agent 连接数据库工具、代码执行工具按用户提问生成分析结果。内部工具串联把公司内部 API 注册成工具让 Agent 按权限调用。但 Agent 不是万能的。它不适合以下场景对实时性要求极高的系统模型推理耗时叠加工具调用耗时可能达到数秒甚至数十秒。完全离线、不允许任何外部 API 调用的内网环境如果用云端大模型数据流转到第三方需要严格评估。高风险决策场景医疗诊断、金融交易等Agent 的幻觉风险不可接受。合规边界必须提前确认所有工具调用要遵循最小权限原则Agent 只能访问它完成任务所必需的资源。涉及用户隐私数据、业务数据的场景必须先获得合法授权且数据处理链路要符合相关法规。不能把 Agent 用于绕过安全限制、攻击系统、窃取账号或批量爬取他人数据。如果 Agent 生成的内容对外发布或商用必须做人工复核。3. 环境准备与前置条件3.1 开发环境检查清单在开始写代码之前先确认以下环境是否就绪项目建议要求说明操作系统Windows 10/11、macOS、Linux核心代码跨平台Python3.10 或更高建议使用 3.11性能和兼容性均衡依赖管理venv 或 conda建议每个项目隔离虚拟环境LLM APIOpenAI 兼容接口的 API KeyDeepSeek、GPT 等均可网络可正常访问 API 服务如果部署本地模型则不受此限制磁盘空间至少 5GB 空闲含依赖和日志空间端口避免 8000、8080、7860 被占用如做 API 服务需提前确认3.2 创建项目目录mkdir agent-toolchain cd agent-toolchain python -m venv venv source venv/bin/activate # Windows 下则执行 venv\Scripts\activate3.3 安装基础依赖pip install openai python-dotenv pydantic requests rich说明openai官方 Python SDK 可以调用 OpenAI 兼容接口DeepSeek 等国产模型也支持同一协议。python-dotenv用于管理环境变量避免把 API Key 写死在代码里。pydantic做数据校验和工具参数 schema 校验。rich用于终端日志输出美化方便观察 Agent 执行过程。4. Agent 工具链的总体架构手写 Agent 工具链最少需要六个模块用户输入 → 规划器 → 工具调度 → 执行器 → 反馈处理器 → 输出结果 ↑______ 记忆模块 ←______|4.1 六个核心模块规划器Planner接收用户目标由 LLM 决定下一步行动。可能拆解为多步任务也可能直接选择工具。工具注册表Tool RegistryAgent 可调用的所有工具的元信息集合。每个工具包含名称、描述、参数 schema、执行函数。LLM 根据描述决定调哪个工具。执行器Executor实际执行工具函数的代码逻辑。注意工具函数在本地运行LLM 只负责给出工具名和参数不直接执行。反馈处理器Feedback Handler把工具执行结果返回给 LLM让模型判断下一步动作。记忆模块Memory Module短期记忆保留当前任务上下文长期记忆按需存储历史事实和偏好。输出解析器Output Parser把 LLM 的响应解析为结构化指令。这是最容易出错的一环也是最需要自定义的地方。4.2 一次完整执行流程第 1 轮用户输入查询北京今天的天气 第 2 轮规划器决定调用 weather.get_current_weather 工具参数 city北京 第 3 轮执行器本地执行天气查询函数返回北京今天晴22 度 第 4 轮反馈处理器将天气结果返回给 LLM 第 5 轮LLM 根据结果生成最终回答这套循环就是 Agent 的“ReAct 模式”Reasoning推理 Acting行动。整个 Agent 引擎的核心就是把这个循环跑稳。5. 从零实现 Agent 核心代码5.1 定义工具基类工具是整个 Agent 的能力来源。设计成装饰器注册模式最灵活# tools.py import inspect from typing import Any, Callable, Dict, List class Tool: def __init__(self, name: str, description: str, func: Callable, parameters: Dict[str, Any]): self.name name self.description description self.func func self.parameters parameters def run(self, **kwargs): return self.func(**kwargs) def to_openai_schema(self) - Dict[str, Any]: return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, } } _TOOL_REGISTRY: Dict[str, Tool] {} def register_tool(name: str, description: str, parameters: Dict[str, Any]): def decorator(func): _TOOL_REGISTRY[name] Tool( namename, descriptiondescription, funcfunc, parametersparameters, ) print(f[Tool Registry] 已注册工具: {name}) return func return decorator def get_tool(name: str) - Tool: if name not in _TOOL_REGISTRY: raise KeyError(f工具不存在: {name}) return _TOOL_REGISTRY[name] def list_tools() - List[Tool]: return list(_TOOL_REGISTRY.values())这里的关键设计点每个工具都有标准化的to_openai_schema()方法可以直接转换成 OpenAI 函数调用协议需要的 JSON Schema。5.2 定义示例工具下面注册两个最常用工具加法和查询时间。# main.py import datetime from tools import register_tool, list_tools register_tool( namecalculator.add, description计算两数相加, parameters{ type: object, properties: { x: {type: number, description: 第一个数字}, y: {type: number, description: 第二个数字} }, required: [x, y] } ) def add(a: float, b: float) - float: return a b register_tool( namesystem.current_time, description获取当前系统时间, parameters{ type: object, properties: {} } ) def current_time() - str: return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) if __name__ __main__: tools list_tools() print(f当前注册了 {len(tools)} 个工具)运行一次确认工具注册正常这是后续所有逻辑的地基。5.3 实现 LLM 客户端封装封装一个ChatClient使用 openai SDK 调用 OpenAI 兼容接口。# llm_client.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() class ChatClient: def __init__(self, model: str deepseek-chat): self.client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.deepseek.com/v1), ) self.model model def chat(self, messages: list, tools: list None) - dict: kwargs {model: self.model, messages: messages} if tools: kwargs[tools] tools kwargs[tool_choice] auto response self.client.chat.completions.create(**kwargs) message response.choices[0].message return { content: message.content, tool_calls: message.tool_calls, }这里有几个细节base_url配置成环境变量方便切换不同模型服务商。tool_choiceauto让模型自主决定调用哪个工具。返回的tool_calls就是模型给出的“工具调用指令”。5.4 实现 Agent 执行循环这是整套工具链最核心的部分。执行循环要做的事发送消息给 LLM。判断是否有工具调用。如果有执行工具将结果追加到消息记录返回 LLM。如果没有工具调用将 content 作为最终回答返回。# agent.py import json from llm_client import ChatClient from tools import get_tool, list_tools class Agent: def __init__(self, system_prompt: str, max_iters: int 10): self.client ChatClient() self.system_prompt system_prompt self.max_iters max_iters self.messages [ {role: system, content: self.system_prompt} ] def run(self, user_input: str) - str: self.messages.append({role: user, content: user_input}) tool_schemas [tool.to_openai_schema() for tool in list_tools()] for step in range(self.max_iters): print(f\n Step {step 1} ) result self.client.chat(self.messages, toolstool_schemas) # 情况一模型没有发起工具调用直接返回结果 if not result[tool_calls]: answer result[content] or 模型未返回内容 self.messages.append({role: assistant, content: answer}) return answer # 情况二模型发起工具调用 assistant_msg {role: assistant, content: None} tool_calls [] for tc in result[tool_calls]: tool_calls.append({ id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, } }) print(f[Agent] 调用工具: {tc.function.name} 参数: {tc.function.arguments}) assistant_msg[tool_calls] tool_calls self.messages.append(assistant_msg) # 逐个执行工具并把结果写回消息记录 for tc in tool_calls: func_name tc[function][name] args_str tc[function][arguments] try: args json.loads(args_str) tool_func get_tool(func_name) output tool_func.run(**args) output_str json.dumps(output, ensure_asciiFalse) except Exception as e: output_str f工具执行出错: {str(e)} print(f[Tool] {func_name} 返回: {output_str[:100]}) self.messages.append({ role: tool, tool_call_id: tc[id], content: output_str, }) raise TimeoutError(f超过最大执行轮数 {self.max_iters} 次任务中止)这个循环已经能跑通最基础的 Agent 能力。注意几个容易踩的坑工具调用后必须追加role: tool的消息并且tool_call_id必须对上。执行工具前要做 try-except防止单个工具报错直接让整个 Agent 崩溃。max_iters必须限制否则模型可能陷入无限循环。5.5 接入记忆模块记忆由两种短期记忆就是上面的self.messages上下文窗口长期记忆需要独立存储。简单实现一个基于 JSON 文件的长期记忆# memory.py import json import os from datetime import datetime class MemoryStore: def __init__(self, memory_file: str memory.json): self.memory_file memory_file self._ensure_file() def _ensure_file(self): if not os.path.exists(self.memory_file): with open(self.memory_file, w, encodingutf-8) as f: json.dump({}, f, ensure_asciiFalse, indent2) def save(self, key: str, value: str): data self._load() data[key] { value: value, updated_at: datetime.now().isoformat() } self._write(data) def load(self, key: str) - str or None: data self._load() item data.get(key) return item[value] if item else None def _load(self): with open(self.memory_file, r, encodingutf-8) as f: return json.load(f) def _write(self, data): with open(self.memory_file, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2)在 Agent 类中把记忆模块组合进去每次对话开始时检查是否有历史的偏好或事实可以注入 system prompt。这里不做太复杂的向量检索先保证“能存能取”。6. Agent 接口 API 与批量任务6.1 将 Agent 封装为 FastAPI 服务工具链搭好之后如果要接业务系统需要把 Agent 暴露成 HTTP API。用 FastAPI 是最快的方式pip install fastapi uvicorn# api_service.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent import Agent app FastAPI(titleAgent Toolchain API, version0.1.0) class AgentRequest(BaseModel): input: str session_id: str default class AgentResponse(BaseModel): output: str session_prompts {} app.post(/agent/run, response_modelAgentResponse) def run_agent(req: AgentRequest): try: if req.session_id not in session_prompts: session_prompts[req.session_id] ( 你是智能助手任务复杂时请拆解为多个步骤使用工具完成。 ) agent Agent(system_promptsession_prompts[req.session_id]) output agent.run(req.input) return AgentResponse(outputoutput) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/agent/tools) def list_agent_tools(): return {tools: [t.to_openai_schema() for t in list_tools()]} if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动命令uvicorn api_service:app --host 127.0.0.1 --port 8000启动后可以直接访问http://127.0.0.1:8000/docs查看 Swagger 接口文档。接口可以跑通后面就可以把 Agent 能力接到自己的业务工具里了。6.2 curl 调用示例curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {input: 请计算 123 456然后告诉我结果, session_id: test-001}预期返回{ output: 123 456 的结果是 579。 }6.3 Python 批量任务调度Agent 服务化之后批量任务就是并发调用的问题。写一个简单的批量调度器# batch_runner.py import json import requests from concurrent.futures import ThreadPoolExecutor, as_completed API_URL http://127.0.0.1:8000/agent/run def run_one(input_text: str, session_id: str) - dict: payload {input: input_text, session_id: session_id} try: resp requests.post(API_URL, jsonpayload, timeout120) resp.raise_for_status() return {ok: True, input: input_text, data: resp.json()} except Exception as e: return {ok: False, input: input_text, error: str(e)} def batch_run(tasks: list[dict], max_workers: int 4): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [ executor.submit(run_one, task[input], task.get(session_id, batch)) for task in tasks ] for future in as_completed(futures): results.append(future.result()) return results if __name__ __main__: tasks [ {input: 计算 1 1, session_id: batch-1}, {input: 计算 10 20, session_id: batch-2}, {input: 计算 100 200, session_id: batch-3}, ] results batch_run(tasks, max_workers3) for r in results: print(json.dumps(r, ensure_asciiFalse, indent2))批量任务的几个要点max_workers不建议开太大因为 LLM API 通常有并发限制且大批量并发会导致延时不稳。每个任务要有独立 session_id避免上下文污染。单个任务必须设置超时时间否则某个任务卡住会占用线程池。批量任务建议先跑 3 到 5 条测试确认接口稳定后再全量跑。7. 资源占用与性能观察7.1 调用云端 API 的资源特点如果你用的是 DeepSeek、GPT 等云端 API本机资源占用非常小主要是 Python 进程内存和网络带宽。一个并发数为 4 的批量调度进程内存通常在 200MB 到 500MB 之间。这种方式的好处是显存无压力劣势是每次调用都要付费、有网络延迟并且存在数据出域的合规风险。7.2 本地部署 LLM 的资源特点如果要完全本地部署需要一个支持 OpenAI 兼容接口的推理服务资源占用取决于模型参数量。一个 7B 参数的量化模型通常需要 6GB 到 10GB 显存14B 模型需要 15GB 以上70B 模型则需要多张显卡或者大内存配合 CPU 推理。实际显存占用必须以你选择的模型和推理框架为准不同量化级别差异很大。7.3 如何观察资源占用开发调试时可以用以下命令观察资源# Linux/macOS top -p pid # Python 脚本内查看内存 import psutil process psutil.Process() print(process.memory_info().rss / 1024 / 1024, MB)7.4 降低资源占用的策略Agent 循环中消息会不断累积建议对历史消息做截断只保留最近 N 轮。工具返回内容如果很大先做截断再塞回上下文。并发批量任务时控制max_workers避免同时创建大量 HTTP 连接。本地部署推理时启用 KV Cache 量化、Flash Attention 等特性但这需要推理框架支持。8. 常见问题与排查方法问题现象可能原因排查方式解决方案调用 API 返回 401API Key 错误或未加载打印环境变量确认.env文件存在重新配置LLM_API_KEYAPI 返回 429触发限流查看 API 返回错误码降低并发数加入重试退避机制Agent 不调用任何工具工具 schema 不合法或模型不认为需要工具打印 tool_schemas 检查格式修正 JSON Schema调低要求显式在 prompt 中提示模型使用工具工具执行报错参数类型不匹配打印 tool_calls 原始 arguments在工具函数内部做类型转换和默认值兜底tool_call_id不匹配assistant 消息和 tool 消息未正确对应检查消息记录顺序严格按“assistant 声明 tool_calls”后逐个追加 tool 响应Agent 陷入无限循环上下文太长或工具返回内容让模型误解查看日志判断重复行为设置 max_iters 上限让工具返回内容更清晰批量任务部分失败网络抖动或 API 限流查看失败任务的错误字段加入超时、重试和失败记录内存持续增长工具执行中对象未释放或消息列表过长监控 RSS 内存变化限制单次任务的消息数及时清理大对象端口被占用8000 被其他服务占用lsof -i :8000或netstat -ano换端口启动8.1 一个典型的 API 重试封装对于 429 和网络错误推荐在调用层加指数退避重试import time import random def request_with_retry(func, max_retries: int 3, base_delay: float 2.0): for attempt in range(max_retries): try: return func() except Exception as e: if attempt max_retries - 1: raise e delay base_delay * (2 ** attempt) random.uniform(0, 1) print(f请求失败{delay:.1f}s 后重试第 {attempt 1} 次: {e}) time.sleep(delay)9. 最佳实践与使用建议9.1 先小后大先慢后快第一次跑通 Agent 时先把 max_iters 设为 3工具只注册一个。等核心循环稳定了再逐步加工具、加并发、加记忆。9.2 保持最小可运行配置建议把“单工具调用”的用例固化为一个自动化测试。每次改动核心循环后先跑这个用例确认没有回归。9.3 工具函数做防御式编程Agent 传入的参数来自 LLM 生成非常不可控。工具函数内部不要信任输入尽量加类型校验、长度限制、空值兜底。比如日期工具要处理非法日期字符串文件工具要限制路径不能越界。9.4 日志和可观测性Agent 的调试难度比普通程序高因为中间多了一层模型推理不确定性。日志至少包含每次 LLM 调用的耗时和 token 数。工具调用名称、参数、返回内容摘要。消息上下文长度变化。错误堆栈和重试记录。建议以结构化 JSON 落盘方便事后排查。9.5 安全边界服务化后不要把 Agent 暴露到公网。如果必须暴露一定要加鉴权。工具的权限限制在最小范围内。比如 Agent 能调用数据库查询工具就只能用只读账号而不是管理员账号。凡是涉及程序执行、文件读写、命令行的工具必须做非常严格的输入校验和沙箱限制。涉及人脸、声音、版权素材、第三方数据的内容需要确认授权后方可使用和分发。9.6 成本控制云端 API 模式下Agent 每轮任务可能调用多次 LLM。一个多步任务可能消耗几千到几万 token。批量任务上线前建议先用一条样本估算 token 消耗。给每个会话设置 token 预算。对过长的工具返回内容做截断。保留每次调用的 token 使用记录按天汇总。10. 总结与下一步这套从零搭建的 Agent 工具链核心价值在于学会了“执行循环”的底层机制。真正把message - tool_calls - tool result - message这条闭环跑通之后你再去使用 LangChain、LangGraph 这类框架会清楚每个抽象层在做的事。先别急着加复杂功能。把最常见的问题验证完单工具调用、多工具连续调用、算数计算、时间查询跑通这四条就说明核心引擎是可用的。最容易踩的坑集中在工具参数解析和消息序列管理上这两个地方多花时间打磨。后面值得继续扩展的方向加入基于向量数据库的长期记忆。加入多 Agent 协作机制比如规划 Agent 和执行 Agent 分离。加入人工确认环节当工具调用风险较高时先暂停等用户确认再执行。加入评测集自动回归测试 Agent 在固定问题集上的稳定性。Agent 开发的核心不在于堆叠多少模型而在于把工具链的每个环节设计得可控、可观测、可恢复。从最小实现开始迭代是能找到瓶颈最快的方式。建议收藏备用动手写第一版循环的时候照着这份架构对照着做会比自己摸索快很多。