自建智能体框架：从调度到工具协议的工程实践

先给结论自建智能体框架不是没有价值而是很多讨论把“要不要自建”和“用不用现成框架”这两个问题混成了一件事。最近只要聊到智能体框架评论区基本都会出现“别重复造轮子”“直接用 LangChain 不香吗”这类声音。这话放在 80% 的普通需求里没错但它忽略了一个关键事实自建框架的价值从来不在于“再写一套 LangChain”而在于把 Agent 的调度逻辑、工具接入、权限边界、批量任务真正变成自己可控的工程资产。这篇文章会从观点、边界、架构、代码、接口、性能、排错几个维度展开。不站“必须自建”的立场也不站在“自建无用”的立场而是把问题拆开看什么场景自建框架是浪费什么场景自建框架是刚需以及如果你决定自建核心模块怎么设计、最小代码长什么样、批量任务和 API 怎么接。1. 核心问题速览问题结论自建智能体框架是不是重复造轮子如果只是把 LLM 调用再封装一层是重复造轮子现成框架能不能覆盖所有场景不能多智能体编排、私有工具接入、权限管控往往是痛点自建框架最大的价值是什么可控的调度逻辑、统一的工具协议、可审计的运行记录自建的成本主要在哪维护成本、架构设计成本、提示词与工具兼容成本什么情况建议自建有私有工具、需要批量任务队列、需要深度定制编排逻辑什么情况不建议自建简单单智能体对话、团队没有工程维护能力、只是为了“不用现成”这个表基本代表了全文立场自建框架有明确的适用边界但“无价值”这个结论过于绝对。2. 先回应“自建智能体框架无价值”的三个论据“自建智能体框架无价值”这个说法通常有三条论据逐条拆一下。第一条论据是“现成框架已经很成熟”。这个说法在一定范围内成立。LangChain、AutoGen、CrewAI、Dify 这些项目在工具调用、记忆管理、多智能体协作上确实沉淀了大量模式普通对话型 Agent 完全可以直接用。但“成熟”和“适合你的场景”是两回事。现成框架的设计目标服务于通用场景一旦你的业务流程里有公司内部的审批接口、私有数据库、专用算法服务通用框架往往需要写大量适配代码这个适配成本并不低。第二条论据是“自建容易写出低质量代码”。这个说法的前提是团队没有足够的架构能力。如果只是把“调用大模型”包成一个类再加一个循环那确实称不上框架也谈不上价值。但判断一个自建框架是否低质量要看它是否解决了真实工程问题。比如是否统一了工具的输入输出协议是否支持批量的失败重试是否记录了一次任务的完整调用链是否能在不修改业务代码的情况下更换底层模型这些能力如果自建代码都不具备那确实该被批评。第三条论据是“社区生态不可替代”。生态确实是现成框架的优势但生态的核心是插件和集成不是框架本身。自建框架同样可以只定义一套工具协议然后把现成框架视为一个可替换的执行引擎。换句话说自建框架不一定意味着放弃生态它可以做在生态之上的一层编排层。所以我的回应观点是自建智能体框架是否有价值取决于你设计它的目的。如果目的是“掌控 Agent 运行的每一个环节”它有价值如果目的是“把简单需求搞复杂”它没有价值。3. 自建与现成框架的边界什么时候选哪个在实际项目里我觉得可以按下面三个维度来决策。第一个维度是需求复杂度。如果你的需求是“给一个文档库做问答”“写一个简单的客服机器人”直接用现成框架最快半天就能跑通。自建在这种场景下是一种消耗。但如果需求里出现了“多个智能体需要按不同角色访问不同数据源”“任务需要排队执行且失败要自动重试”“输出结果要进入内部审核流程”这些复杂协作自建的价值就会凸显出来因为你需要的是对流程的精确控制。第二个维度是工具的私有程度。企业项目里真正麻烦的不是调用 OpenAI 或国内大模型 API而是那些只有内部系统才有的接口。通用框架能帮你调用一个 HTTP API但它很难替你设计“工具结果需要脱敏”“工具调用需要走内部审批”“工具返回格式需要统一转换”这些策略。这些策略恰恰是自建框架的强项你可以把工具接入做成一套插件协议每个工具只负责实现标准接口其他逻辑由框架统一处理。第三个维度是团队的维护能力。自建框架一旦开始就是长期维护的资产意味着每次大模型 API 更新、新增工具、调整调度策略都要有人跟。如果团队只有一两个人且项目周期短不要自建。如果项目是长期业务系统的一部分自建框架的维护投入可以被摊薄这时候才值得做。用一个不太严谨但直观的判断方式如果业务逻辑里有一半以上的“编排规则”是现成框架不支持的那自建就是合理的如果现成框架只差一个工具调用那写个适配层就够了。4. 自建智能体框架的核心架构设计如果决定自建首先要明确一点智能体框架的本质不是“调用大模型”而是“管理调用大模型前后的所有事情”。一个最少可用的自建框架至少需要下面 5 个模块。4.1 调度器调度器是框架的大脑。它负责接收任务、决定任务由哪个 Agent 执行、检查执行结果、处理失败重试、控制最大轮数。调度器的设计重点是把“任务”抽象成数据结构而不是在代码里写一堆 if else。任务至少应该包含任务 ID、输入文本、关联工具列表、最大执行轮数、状态字段。4.2 工具协议自建框架最值得花时间设计的就是工具协议。推荐做法是每个工具实现一个标准的函数签名输入是 JSON 字符串或字典输出也是结构化数据。框架不关心工具内部怎么实现只负责把大模型给出的工具调用意图路由到对应工具然后把工具返回结果回传给大模型。# 工具协议示例所有工具函数统一接收 dict返回 dict def search_order(query: dict) - dict: order_id query.get(order_id) # 内部调用订单系统接口 return {success: True, data: {order_id: order_id, status: shipped}}这种协议的好处是新增工具时不需要改主循环只需注册函数名和描述。4.3 记忆模块记忆模块决定了智能体能不能在多轮对话里保持上下文一致性。自建框架不一定要做向量数据库可以先从最简单的会话历史列表开始把每一轮的 user 消息、assistant 消息、工具调用结果按顺序拼接。当上下文长度到达阈值时做截断或摘要。这个模块初期不需要复杂但要预留存储接口方便后续接数据库或向量库。4.4 模型适配层模型适配层负责屏蔽不同大模型 API 的差异。做法是定义一个统一的 chat 接口返回统一的响应结构然后为不同模型写适配器。这样业务代码不直接依赖任何具体模型服务换模型时只换适配器。class BaseLLM: def chat(self, messages: list[dict]) - str: raise NotImplementedError class OpenAICompatLLM(BaseLLM): def __init__(self, base_url, api_key, model): self.base_url base_url self.api_key api_key self.model model def chat(self, messages: list[dict]) - str: # 调用 OpenAI 兼容接口 # 实际路径、请求头需按服务商文档调整 return response_text4.5 日志与观测这是最容易忽略但最关键的模块。自建框架必须记录每一次任务的完整调用链输入是什么、调用了哪些工具、每个工具返回了什么、大模型最终输出是什么、一共执行了几轮、耗时多少。没有这套日志你根本没法排查“AI 为什么答错了”也没法评估框架的稳定性。5. 最小可运行的自建 Agent 框架示例这里给一个最小可运行的智能体框架核心代码。它是一个“工具调用循环”大模型根据用户的自然语言任务决定是否调用工具、调用哪个工具、传什么参数然后框架执行工具并把结果返回给大模型直到大模型认为不需要再调用工具为止。import json import uuid class SimpleAgent: def __init__(self, llm, toolsNone, max_iterations5): self.llm llm self.tools tools or {} self.max_iterations max_iterations def _tool_descriptions(self) - str: lines [] for name, func in self.tools.items(): lines.append(f- {name}: {func.__doc__}) return \n.join(lines) def _call_tool(self, name: str, args: dict): tool self.tools.get(name) if not tool: return {success: False, error: ftool {name} not found} try: return tool(args) except Exception as exc: return {success: False, error: str(exc)} def run(self, task: str) - dict: if not self.tools: result self.llm.chat([{role: user, content: task}]) return {task_id: str(uuid.uuid4()), result: result, steps: []} system_prompt ( 你是智能体调度器。你可以使用以下工具\n f{self._tool_descriptions()}\n 如果任务需要调用工具请输出 JSON格式为 {tool: 工具名, args: {}}。 如果任务已经完成请直接输出最终回答不要输出 JSON。 ) messages [{role: system, content: system_prompt}, {role: user, content: task}] steps [] for _ in range(self.max_iterations): response self.llm.chat(messages) messages.append({role: assistant, content: response}) if not response.strip().startswith({): return {task_id: str(uuid.uuid4()), result: response, steps: steps} try: call json.loads(response) tool_name call.get(tool) args call.get(args, {}) except json.JSONDecodeError: return {task_id: str(uuid.uuid4()), result: response, steps: steps} tool_result self._call_tool(tool_name, args) steps.append({tool: tool_name, args: args, result: tool_result}) messages.append({role: tool, content: json.dumps(tool_result, ensure_asciiFalse)}) return {task_id: str(uuid.uuid4()), result: reach max iterations, steps: steps}这个代码虽然简陋但它已经具备了自建框架最基本的形态任务 ID、工具注册、工具结果回填、最大迭代限制、调用步骤记录。你可以在这个基础上扩展出多 Agent、任务队列、权限校验、结果审核等能力。测试这个最小框架只需要两步首先注册一个或多个模拟工具然后传入一个需要调用工具的任务。def get_weather(args: dict) - dict: 查询城市天气。参数 city 为城市名。 city args.get(city, ) return {success: True, data: {city: city, weather: 晴, temperature: 26}} tools {get_weather: get_weather} agent SimpleAgent(llmyour_llm_instance, toolstools) result agent.run(北京今天需要带伞吗先查一下北京的天气) print(result[result])从材料看如果你只是想快速跑通一个原型不必一开始就追求复杂设计先让主循环稳定再逐步加模块是更稳妥的路径。实际部署时的模型接入路径、推理参数、请求超时设置都要以你接的模型服务商文档为准这里只给框架层面设计参考。6. 接口 API 与批量任务设计自建框架要真正进入生产环境必须解决两个问题对外提供 API、对大量任务做批量处理。6.1 API 服务推荐用 FastAPI 把 Agent 封装成 HTTP 接口这样可以和现有业务系统对接。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(title自建 Agent 框架 API) class TaskRequest(BaseModel): task: str agent_name: str default class TaskResponse(BaseModel): task_id: str result: str app.post(/agent/run, response_modelTaskResponse) async def run_task(req: TaskRequest): agent get_agent(req.agent_name) if agent is None: raise HTTPException(status_code404, detailfagent {req.agent_name} not found) result agent.run(req.task) return TaskResponse(task_idresult[task_id], resultresult[result])启动命令参考uvicorn api_server:app --host 0.0.0.0 --port 8000然后可以用 curl 验证接口curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {task: 查一下北京天气, agent_name: default}要注意接口服务一旦对外开放必须加访问控制和请求体大小限制避免被刷。6.2 批量任务批量任务的核心是一个任务队列。最简单的设计方案是用 Python 的 queue 多线程但生产环境建议用 Redis 或数据库表做持久化队列这样服务重启后任务不会丢。下面是批量任务的最小实现思路import queue import threading import time task_queue queue.Queue() def worker(agent): while True: task task_queue.get() try: result agent.run(task) save_result(result) except Exception as exc: save_error(task, exc) finally: task_queue.task_done() def start_workers(agent, worker_count3): for _ in range(worker_count): t threading.Thread(targetworker, args(agent,), daemonTrue) t.start()批量任务设计要重点关注三个点任务状态、失败重试、结果落盘。任务至少要有 pending、running、success、failed 四种状态。失败任务建议进入重试队列重试次数限制在 2 到 3 次超过则标记失败并写入错误原因。7. 资源占用与性能观察智能体框架的资源消耗主要来自两个部分底层 LLM 推理服务以及框架自身的运行开销。如果底层模型是部署在本地的显存占用由模型大小和推理参数决定这部分与框架本身无关。之前见过一些本地部署的智能体工程7B 级别模型量化后在 6G 到 8G 显存左右可以跑起来但具体占用要按实际模型版本、量化方式、上下文长度、并发数来观察不能拿别人参数直接套。如果调用的是云端 API框架自身的资源占用非常低主要是 CPU 和内存。一个简单的 FastAPI 服务加上队列内存占用通常在几百 MB 级别。但要注意并发量上来之后日志存储、任务结果存储、向量检索这些配套组件会成为新的资源瓶颈。7.1 性能观察方法观察框架性能推荐从这几个指标入手单任务平均耗时从收到任务到返回最终结果的完整耗时。工具调用次数一个任务平均触发多少次工具调用次数越多耗时越高。大模型 API 超时率调用底层模型时的超时比例过高说明并发配置不合理。队列积压数批量任务模式下队列中等待执行的任务数量。日志存储增速完整调用链日志会占磁盘需评估长期存储成本。7.2 降低开销的手段任务并行度要控制。多线程不是越多越好底层模型并发能力有限盲目加大 worker 数量只会提高 API 超时率。减少无效工具调用。大模型经常在信息已经足够时还继续调用工具可以在提示词里约束“完成就停止”。设置合理的最大迭代次数。10 轮上限比 5 轮上限更容易出复杂结果但也会显著增加单任务耗时长尾。在自建框架没有观察到真实压力测试前不要轻易相信“我的框架支持高并发”这种结论先用小并发压测再说。8. 常见问题与排查方法自建智能体框架的故障点通常不在框架本身而在模型调用、工具返回、任务队列这些衔接处。问题现象可能原因排查方式解决方案Agent 不调用工具提示词里工具描述不清晰或模型不支持 function calling查看单轮日志中模型输出调整提示词明确工具触发条件工具调用参数错误工具描述没有写清楚参数格式检查模型输出中的 JSON 参数在工具描述里增加参数示例任务一直循环不结束最大迭代次数设置过高模型反复调用工具查看 steps 日志降低 max_iterations增加“完成就停止”提示API 接口返回超时底层模型推理时间过长HTTP 超时设置太短观察单任务平均耗时增大超时时间或改为异步任务模式批量任务卡在 runningworker 崩溃或队列消费异常查看 worker 日志增加 worker 异常捕获标记失败任务换模型后效果不稳定不同模型的 JSON 输出格式有差异对比模型原始返回在模型适配层做输出格式清洗日志磁盘增长过快每次任务记录全量调用链查看日志目录大小设置日志轮转归档旧任务记录并发高时 API 报错底层模型限流查看错误码增加任务排队降低 worker 并发排查自建框架问题最重要的原则是先分环节问题出在用户输入、大模型输出、工具执行、还是结果回传把一次任务的完整日志打印出来90% 的问题都能定位到。9. 最佳实践与合规要求自建框架不意味着可以忽略规范和审计。这里按工程实践和合规两条线说明。工程实践上第一次先把单 Agent 单工具跑通不要一上来就设计多智能体协作。建议保留一套最小可运行配置包括一个可用的模型 API Key、一个模拟工具、一份示例任务方便回归测试。工具和 Agent 配置要独立于业务代码尽量用 JSON 或 YAML 文件管理调参。批量任务一定要加日志、状态和失败重试不能把任务直接抛给一个裸线程池。API 服务要限流可以用简单的 IP 白名单或请求频率控制。输出结果在进入业务系统前要做格式校验不能直接把大模型输出当结构化数据用。安全与合规边界上需要注意如果工具涉及内部系统、用户数据、订单信息必须做权限校验不能因为“方便测试”就放开访问。涉及人脸、声音、肖像或版权素材的任务必须有明确授权才能处理这在智能体接入图像或音视频处理工具时尤其重要。大模型输出的内容在发布前要做人工复核自动生成的内容不能直接对外发布。涉及模型使用条款和数据处理范围时要按照模型提供方的使用规范操作。自建框架更容易被忽略的问题是“越权”一个工具只管执行但谁有权触发这个工具、工具返回的数据谁能看到这些判断必须由框架统一控制。不要在工具内部各自实现权限判断那样会失控。10. 总结与下一步聊回最初的问题自建智能体框架到底有没有价值。我的判断是如果你的系统里只有一个“用户提问模型回答”的链路自建框架没有价值现成工具更省事。但如果你的业务里存在私有工具、批量任务、权限边界、多步骤调度这类需求自建一个足够薄的框架是非常划算的投入。它的价值不在于“写了很多代码”而在于把不可控的模型输出变成了可观测、可重试、可审计的工程流程。最值得先验证的功能是工具协议先让一个 Agent 稳定调用一个模拟工具观察提示词设计和 JSON 回传是否稳定再考虑扩展。最容易踩的坑是把自建框架做成一个大而全的平台第一版就追求多智能体、知识库、可视化编排结果连最基础的工具调用循环都没跑稳。后续可以扩展的方向包括把队列换成 Redis 持久化、给工具调用加权限校验中间件、引入向量数据库做长期记忆、在框架层面接入可观测平台等。如果你正在做一个长期存在的业务系统自建智能体框架解决的不只是“能用”而是“可控”。这套思路整理下来可以直接作为你自建智能体框架的第一版设计草稿建议收藏备用。