自建AI Agent框架：核心组件设计与实战踩坑指南

做AI Agent开发这段时间，我一直围绕自建的hello-agents框架打转。说实话，Agent框架搭建这件事，看着简单，真正跑起来全是细节。这是《探秘 AI Agent | Hello-Agents 项目学习笔记》的第六篇，前五篇我记录了从环境准备到模型接入的完整过程，这一篇不想再聊具体API，而是想把Agent框架搭建这件事拆到根上：为什么自建、核心组件怎么设计、怎么用最少的代码跑通一个可用Agent、以及长期困扰人的坑该怎么填。这篇笔记适合已经跑过几个LLM API调用，但不满足于“调库”的人。如果你正在纠结LangChain、Dify、CrewAI到底选哪个，看完这篇，你可能会对“自建”有不一样的理解。

1. 为什么自建Agent框架：主流框架没说的另一面

1.1 主流Agent框架的舒适区与边界

过去半年，我把市面上能叫得上名字的Agent框架都试了一遍，包括LangChain、Dify、CrewAI、Spring AI，甚至用Rust写过几个小型Agent项目。每个框架都有自己的高光时刻，但也都有明显边界。

这些框架解决的是80%的通用问题，剩下20%的“不通用”需求，恰恰是项目里最挠头的地方。比如我要精确控制某个Agent的上下文窗口，或者想在每次模型调用前统一注入业务侧变量，主流框架并不会把这种底层逻辑直接暴露给我。再比如token成本，很多框架把历史消息自动塞给你，你无感知地多花了钱，出了问题都不知道是框架哪一层干的。

自建框架的意义，不是否定这些项目，而是把黑盒打开看一眼。我个人的体会是：用LangChain三个月，不如手写一个最小Agent循环来得通透。

1.2 Hello-Agents框架的设计定位

hello-agents是我自己维护的一套轻量Agent框架，定位很明确：不追求大而全，只保留Agent运行最必要的能力。它由五个模块构成：模型接入层、工具注册表、记忆管理、Agent执行循环、编排器。

设计初衷源于一个很朴素的诉求——我想知道一条用户消息从进入系统到模型返回，中间到底发生了什么。用别人框架时，我只能通过日志猜测；用自建框架时，每一条消息都是我写的代码在处理。代码即文档，出了问题翻代码比翻文档快得多。

所以hello-agents的核心理念就是“最小可运行、显式优于隐式”：一个Agent对象，一个agent.run()方法，中间的过程参数尽量显式暴露出来，哪怕多写几行代码，也不搞暗箱操作。

1.3 自建框架的收益与代价

收益是最直接的：第一，完全可控，模型调用、工具执行、异常处理全部在自己手里；第二，深入理解，Agent框架最核心的ReAct循环、Function Calling、记忆管理，手写一遍之后，再去看LangChain的AgentExecutor源码会轻松很多；第三，方便裁剪，想支持CrewAI那种多角色协作，自己加一个编排层就行，不需要被框架的规范束缚。

代价也很诚实：需要自己处理模型API差异、超时重试、并发限流、token记账，这些脏活累活，主流框架已经帮你蹚过一遍了。如果你只是快速搭建一个Demo，自建框架的性价比不高；但如果你要长期维护一个Agent系统，这些“脏活”迟早都得懂。

我的建议是：可以先自建一个最小框架做学习或内部工具，再根据业务成长，选择保留自建还是迁移到成熟框架。hello-agents目前就是我这个思路的产物，它还很糙，但足够真实。

2. Agent框架搭建的核心组件：一个也不能少

2.1 Agent运行循环：五步理解ReAct

Agent框架最核心的部分是执行循环，业界叫ReAct模式，网上常说的“react agent框架图”其实就是下面这个循环：

1. 接收任务和系统提示词
2. 模型推理（Thought）：决定下一步做什么
3. 调用动作（Action）：选择并调用工具
4. 观察结果（Observation）：拿到工具返回信息
5. 判断是否结束：给不出最终答案就继续循环

用伪代码表达，一个最小Agent循环只有十几行：

def run(task: str): messages = [system_message, user_message(task)] for step in range(max_steps): response = llm.chat(messages) if response.is_final_answer: return response.content action = response.tool_call observation = execute_tool(action) messages.append(observation_message(observation)) return "MAX_STEPS_REACHED"

这段逻辑看起来简单，但整个框架搭建的复杂度都藏在这个循环的细节里：解析模型输出、匹配工具参数、处理工具异常、控制上下文长度、避免死循环。

2.2 模型接入层与token管理

很多初学者问“ai agent token是什么意思”，在Agent框架里，token就是模型处理文本的最小单位，你的每一次推理、每一个工具调用结果、每一段历史消息，最后都会折算成token计费。

模型接入层要做的第一件事，是把OpenAI、Gemini、Claude这些厂商的差异“抹平”。在我自建的hello-agents里，定义了一个ChatModel接口，统一接收消息列表和参数，返回标准化的模型响应。底层再接不同厂商的SDK，上层不用关心。

token管理则是另一件容易被忽视的事。Agent每多跑一轮，历史消息就会膨胀一轮，成本是线性甚至指数上升的。所以hello-agents做了一个上下文管理器，负责三件事：统计当前轮次的输入和输出token，估算费用；超出窗口时自动裁剪最早的历史；对长工具结果做摘要压缩。实际测试下来，同样一个检索类任务，有了这层管理后token消耗能下降30%到50%。

2.3 工具层与Function Calling

如果说Agent循环是骨架，工具层就是血肉。hello-agents里，工具的注册不是简单定义函数，而是要生成一份模型可以理解的JSON Schema。以OpenAI Function Calling为例，你希望Agent调用一个搜索文档的工具，需要这样注册：

@tool("search_docs") def search_docs(query: str, max_results: int = 5) -> str: """根据关键词搜索项目文档，返回匹配内容""" # ...实际检索逻辑

框架会解析这个函数的签名、类型、docstring，自动转成tools参数传给模型。模型看到这个schema后，如果发现需要检索，会返回一个结构化的调用请求：

{ "name": "search_docs", "arguments": "{\"query\": \"Agent框架搭建\", \"max_results\": 3}" }

框架要做的是把arguments解析成Python参数，调用函数，把结果追加回对话。这个链路看着不复杂，但它决定了Agent能不能“下地干活”。我在实际使用中强烈建议：工具函数的docstring一定要写清楚“什么时候用”以及“参数边界”，模型对工具的误判，80%都是因为说明书写得含糊。

2.4 记忆与状态管理

Agent框架搭建里，记忆模块最容易被新手跳过，但恰恰是它决定了Agent是“有脑子的助手”还是“每句话翻篇的鹦鹉”。Hello-agents把记忆分成两层：短期记忆和长期记忆。

短期记忆直接复用对话上下文，靠消息列表携带，核心是控制长度与相关性。长期记忆则会抽取出关键实体、偏好、历史决策，存入向量库，在每次任务开始时检索与当前问题相关的片段，作为额外上下文注入。

这里有一个很大的坑：状态管理的边界。如果Agent在执行多步工具调用时，某一步突然抛异常，整个状态是应该回滚、重试，还是放弃？我最初的实现是直接抛出异常，结果经常导致Agent任务中断。后来我改成在循环内捕获异常，把错误信息当作一次observation返回给模型，让模型自行判断下一步。这个改动让任务成功率明显提升，因为模型往往能从错误信息里找到原因。

2.5 编排层：单Agent到多Agent协作

自建框架做到第二个版本，你就会发现单Agent能力存在天花板：既要做分析、又要写代码、还要跑测试，上下文很快被占满，工具切换也容易出错。这时候需要引入编排层，也就是让多个Agent各司其职、协同工作。

CrewAI的“角色-任务-流程”思想值得借鉴。我把hello-agents的编排器设计成一张有向图，节点是不同Agent，边是消息传递关系。比如一个简单的“需求分析流程”，可以定义需求Agent先解析需求，产出结构化任务清单；然后开发Agent消费任务清单，产出代码；最后测试Agent消费代码，产出测试报告。

这个设计比单Agent循环复杂，但本质逻辑一致：每个子Agent都有自己的system prompt、工具集合、循环上限，编排器只负责调度和传递上下文。先别急着加多Agent，单Agent跑不稳，多Agent只会把错误放大三倍。

3. 实操：基于Hello-Agents搭一套可运行流程

3.1 目录结构与接口设计

hello-agents目前的目录结构是这样拆的：

hello_agents/ ├── agent.py # Agent核心类与执行循环 ├── model.py # 模型接入统一接口 ├── context.py # 上下文与token管理 ├── memory.py # 短期/长期记忆 ├── tools/ │ ├── registry.py # 工具注册表 │ └── builtin.py # 内置常用工具 ├── orchestration.py # 多Agent编排器 └── server.py # FastAPI对外服务

这种划分的核心原则是“单一职责”：模型、工具、记忆、执行、编排互不感知实现细节，只通过接口通信。好处是换一个模型接入、加一个新工具，都不用动Agent循环。坏处是初期要多写两层包装代码，不过这个付出很值得。

3.2 手写ReAct循环：先让框架跑起来

下面这段代码是hello-agents最早的Agent核心，精简后大概五十行，但它足以跑通一个基础的Agent流程。实现思路参考ReAct循环，但我在里面加了三个关键控制：最大步数、重复检测、异常恢复。

class Agent: def __init__(self, model, tools, max_steps=8): self.model = model self.tools = tools self.max_steps = max_steps self.history = [] def run(self, task: str) -> str: self.history = [system_message, user_message(task)] last_observations = [] for step in range(self.max_steps): response = self.model.chat(self.history) if response.type == "final": return response.content tool_name = response.tool_call.name arguments = json.loads(response.tool_call.arguments) try: observation = self.tools.execute(tool_name, arguments) except Exception as e: observation = f"工具执行失败: {type(e).__name__}: {e}" # 重复检测：如果连续三次出现相同工具和相同参数，直接打断 last_observations.append((tool_name, arguments)) if len(last_observations) >= 3 and len(set(map(str, last_observations[-3:]))) == 1: return "Agent陷入重复调用，已中断" self.history.append(assistant_message(response.content, tool_call=response.tool_call)) self.history.append(tool_message(observation, name=tool_name)) last_observations.append(observation) return "超过最大执行步数"

这段代码最重要的不是实现本身，而是它揭示了一个事实：Agent框架搭建的核心不是炫技，而是把“循环控制”做好。你在生产环境里遇到的大部分问题，无外乎循环出不去、重复执行、工具调用失败，只要在循环上补好刹车，框架就已经及格了一半。

3.3 让Agent下地干活：工具接入与安全边界

框架跑通后，就要让它真的“下地干活”。我接的第一个实用工具是内部文档搜索，第二个是SQL查询库。这两类工具都有安全隐患，所以框架必须做隔离和限制，我总结了三条铁律：

1. 工具必须显式注册，不允许Agent动态生成工具函数。
2. 所有工具执行前要校验参数类型和取值范围，比如SQL工具只允许SELECT，不允许DELETE。
3. 工具执行要有超时控制，默认10秒，超时直接返回错误，不让模型挂在那里干等。

实际运行的坑在于模型经常会产生“幻觉参数”。比如文档搜索工具支持日期范围，模型会自作主张把当前日期当参数传进去，而框架记录里根本没有这个值。后来我加了参数过滤，对于模型未提供的参数，一律使用工具的默认值，不进行二次推断。

安全边界上还要注意：Agent可以调用工具拿到真实数据，但最终对外输出前应该有一个审核层。尤其是涉及企业内部数据的场景，不能盲目相信模型生成的内容。hello-agents目前的做法是给所有工具返回的数据打上“数据来源”标签，后续可以做引用溯源。

3.4 从同步到异步：并发与稳定性改造

最早版本的hello-agents是同步实现，一次只能处理一个任务，慢得像单线程隧道。后来我接了一个小工具，需要给业务部门批量处理几十条文本，才意识到“ai agent怎么扛并发”这个问题有多迫切。我的改造方向是asyncio加信号量：

import asyncio from asyncio import Semaphore async def run_many(tasks, model, tools, limit=5): sem = Semaphore(limit) async def bounded(task): async with sem: agent = Agent(model=model, tools=tools) return await asyncio.to_thread(agent.run, task) # 同步循环放入线程池 results = await asyncio.gather(*[bounded(t) for t in tasks]) return results

这里有三个容易被忽略的点。

第一，很多模型SDK的聊天接口是同步阻塞的，直接放在async事件循环里会把整个循环卡死。我的做法是用asyncio.to_thread把同步逻辑丢到线程池，保持接口响应不阻塞。

第二，并发上限不是越高越好，要参照模型API的限流额度。我试过把并发调到20，结果一分钟后被限流，后面全是429。正确姿势是先查API文档确认RPM和TPM限制，再设置合理的semaphore。

第三，并发场景下要做好请求级别的隔离。每个任务的history不能共享，模型实例也不能共用，否则一个任务里的上下文会污染另一个任务。我在实际开发中因为这些共享变量查了一整天才定位到问题。

3.5 接入FastAPI对外服务

并发改造完之后，下一步就是提供服务。hello-agents的对外服务我用的是FastAPI，因为它和asyncio天然契合，而且自带请求校验和文档。核心接口就两个：提交任务、查询任务结果。

由于Agent任务执行时间可能很长，所以没有用同步请求等待结果，而是采用异步任务队列：收到HTTP请求后创建任务ID，返回202；后台执行Agent循环，执行完成后把结果写入内存或Redis；前端轮询结果。这一步可以参考FastAPI的BackgroundTasks，也能用Celery做更可靠的任务分发。

贴一个最简化版的服务代码：

from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app = FastAPI() tasks_store = {} class TaskIn(BaseModel): content: str @app.post("/agent/task") async def create_task(payload: TaskIn, background_tasks: BackgroundTasks): task_id = str(uuid.uuid4()) tasks_store[task_id] = {"status": "running", "result": None} background_tasks.add_task(execute_bg, task_id, payload.content) return {"task_id": task_id} async def execute_bg(task_id, content): result = await agent.run_async(content) tasks_store[task_id] = {"status": "done", "result": result}

实际部署时，还可以在FastAPI前加一层简单限流中间件，防止外部请求把模型额度打爆。这个服务上线后，我自己的体验是：API调用方不用关心Agent内部有多少步，只需要拿到任务ID和最终结果，调用关系清爽很多。

4. 常见问题与排查技巧实录

4.1 Agent陷入无效循环怎么破

Agent框架搭建中最常见的问题是：模型在循环里反复调用同一个工具，或者答非所问。排查思路分三步。

第一步，确认步数上限有没有设置。很多“无限循环”其实是忘了配max_steps，模型为了凑最终答案会一直尝试。

第二步，观察连续几步的observation是否高度相似。如果是，大概率是模型没有从工具返回中得到新信息，可能原因有：工具返回的内容太冗余，模型被噪声淹没；或者工具查询参数本来就有问题，结果每次返回都是同一批数据。

第三步，检查输出格式解析。OpenAI模型偶尔会在函数调用之外额外输出解释性文本，如果框架的解析逻辑只认严格JSON，就会出现“解析失败->重试->又失败”的死循环。解决办法是增加容错解析，先尝试JSON解析，失败就用正则提取参数块。

我在hello-agents里还加了一个特殊机制：如果检测到“工具名+参数”的组合连续出现三次，就判定为无效循环，主动终止并返回“当前Agent无法解决该任务”。宁可让它承认失败，也不让它烧掉几十轮token。

4.2 token消耗像流水一样，怎么控

token失控是最让团队肉疼的问题。一个看起来简单的任务，可能背后跑了十几轮工具调用，每个工具结果都带着几千字上下文返回，最后账单出来吓一跳。这里的核心不是换便宜模型，而是控制上下文增长。

我现在的做法有三个：历史消息压缩，超过3轮对话就触发一次摘要，用摘要替换最早的具体内容；工具结果截断，每个工具返回只保留前2000字符，超出部分加一个“内容过长，已截断”的标记；限制并行推理轮数，如果任务没有新的上下文输入，模型调用的步数上限可以收紧。

还要特别关注“隐式token开销”。很多框架会在系统提示词里塞一大堆工具说明，每个工具的schema都完整展开，这部分token虽然看不见，但每次请求都在消耗。hello-agents做了工具按需挂载：根据任务关键词预选出可能用到的三个工具，只给模型暴露这三个工具的schema，实测可减少20%左右的输入token。

4.3 并发场景真的扛不住吗

当你把Agent接到真实业务里，“ai agent怎么扛并发”就会变成一个避不开的话题。但我的真实感受是：多数场景下的瓶颈不在Agent框架本身，而在模型API的限流。框架再怎么优化，也突破不了“每秒请求数”和“每分钟token数”这两个硬限制。

所以扛并发的核心是“削峰填谷”：用队列缓冲请求，用并发控制限制同时打到API的请求数，用退避重试处理429和超时。具体参数可以参考这张速查表：

如果你的项目对延迟极度敏感，并发冲击又大，我可以坦诚地说，Python Agent在高并发下的性能上限不高。这时候可以考虑把Agent核心循环迁移到Rust，或者至少把工具调用、上下文管理这种重计算模块用Rust重写为原生扩展，我做过一个实验，单Agent循环在Rust下的开销比Python低一个数量级，当然开发时间也成倍增加。

4.4 和LangChain、Dify、CrewAI对比后的选择建议

这段时间反复折腾自建框架，我对“agent框架如langchain、dify、crewai等，哪个好”这个问题的回答已经变了：先看你的需求是“要一个系统”还是“要一个自己掌握的系统”。

如果是业务要快速跑通一个问答机器人，Dify的可视化编排最合适；如果团队已经深度使用LangChain的生态，比如大量用它的文档加载器和向量库，那就继续用LangChain，没必要推倒重来；如果重点是多个角色协作，CrewAI的封装能省很多事；如果你在Java技术栈里，Spring AI是稳妥选择。

反过来说，如果你需要反复调整Agent的决策逻辑、精细控制token成本、或者把Agent嵌入到一个定制化很强的业务系统里，自建框架反而更顺手。hello-agents和它们的最大区别是：我对每一行执行代码都了如指掌，出了问题能直接修，而不是等框架作者发版本。

这里还想补充一句：不要把“自建”理解成“从零造轮子”。我在hello-agents里也借鉴了很多成熟框架的思路，比如CrewAI的编排思想、LangChain的工具Schema设计，复用思想不丢人，丢掉思考才可惜。

4.5 后续扩展方向

hello-agents目前的版本还处于“个人生产可用”阶段，接下来我计划做四件事。

第一，更完善的多Agent编排，目前的编排器只支持简单链式调用，下一步会支持并行执行和条件分支。第二，增加多模态输入支持，让Agent能处理图片和语音，这需要模型接入层做一轮重构。第三，把核心执行循环用Rust重写为网络服务，Python侧只做调度，解决性能焦虑。第四，补充更细粒度的审计日志，每个Agent每轮的token消耗、工具调用、决策原因都能可视化回溯。

如果你也是自建Agent框架，我建议从“给框架写文档”开始。hello-agents每新增一个能力，我就先写设计文档，再说代码。这套流程意外地好用，因为写文档的过程会逼你想清楚这个模块到底为什么存在。

最后分享一个很个人的体会：Agent框架搭建，一半是工程，一半是认知。工程上无非是模型、工具、记忆、循环、编排这几件事，认知上却要不断在“自主性”和“可控性”之间找平衡。框架太激进，Agent会像脱缰的野马乱跑；框架太死板，Agent又失去了智能决策的意义。hello-agents目前的答案很简单：框架只做流程控制，不限制模型思考；工具只做能力边界，不干预模型判断。你先把这个平衡找到，再谈复杂功能也不迟。