资讯动态

Agent内核设计实战:从零实现可运行的决策循环与工具注册框架

发布时间:2026/9/8 9:34:00 来源:尧图企业网站定制
Agent 开发入门时很多人会先选一个现成框架跑通一个 Demo 后感觉“我也会做 Agent 了”。但一旦进入真实业务要调整推理流程、控制工具调用权限、设计多轮记忆策略时才发现自己根本改不动框架只能被框架牵着走。这就是“会用框架”和“能设计内核”之间的差距。这篇文章想拆解的是 Agent 内核设计中最关键的部分决策循环、工具注册、记忆管理和上下文组装。我会以 DeepSeek-Honeycomb 这个自研 Agent 项目的源码结构为线索讲清楚一个 Agent 内核到底由哪些模块组成、每个模块为什么存在、模块之间怎么协作并给出一个可运行的最小内核骨架。读完你至少能回答三个问题Agent 内核和普通业务代码的本质区别是什么一个最小可用内核需要哪些核心抽象如果自己要写一个轻量 Agent 框架第一行代码应该从哪里开始。1. 为什么要读 Agent 内核源码从“跑通 Demo”到“能改框架”先抛一个判断Agent 开发的第一道分水岭不是会不会调 Prompt而是能不能理解并改造内核里的主循环。很多团队开发 Agent 应用时遇到的第一堵墙不是大模型能力不够而是框架约束太多。你需要在每次调用模型前注入系统权限规则框架不开放这个 Hook你需要让 Agent 在工具调用失败后自动重试并修正参数框架只做了单次调用你需要把历史对话压缩成摘要而不是全量拼接框架的 Context 管理策略是固定的。这些需求如果发生在业务代码里你可能直接改业务逻辑就完了。但发生在框架层你没有源码级的理解就只能 hack而 hack 的代码永远追不上框架升级。DeepSeek-Honeycomb 这类自研 Agent 项目之所以值得拆解是因为它没有把 Agent 做成一个不可分割的“黑盒应用”而是把内核拆成了可替换的模块。理解这种模块化设计比记住某个框架的 API 更值钱因为底层原理是通用的。读完这节你应该建立的认知是Agent 内核的源码价值不在于代码本身多复杂而在于它展示了“LLM 应用如何被组织成一个可控制的循环”。2. Agent 内核到底是什么核心概念与架构分层2.1 从一次普通 API 调用说起先看一段最普通的 LLM 调用代码# 文件路径demo/simple_llm_call.py from openai import OpenAI client OpenAI() response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是一个有用的助手}, {role: user, content: 今天天气怎么样} ] ) print(response.choices[0].message.content)这段代码的问题在于它是一次性的、无状态的、没有工具能力、没有记忆能力。模型回答完进程结束什么都不留下。2.2 Agent 内核多了什么Agent 内核和普通 LLM 调用的本质区别可以用一张表说清楚维度普通 LLM 调用Agent 内核调用次数一次多次直到任务完成或达到上限状态管理无维护会话上下文、中间结果工具使用无有 ToolRegistry可注册、可调用环境交互无通过工具读取外部世界并施加影响终止条件模型返回即结束由终止策略决定如最大轮次、目标达成记忆策略全部拼进 Prompt可裁剪、可摘要、可长期持久化所以 Agent 内核本质上是一个“模型调用 工具调用 状态管理”的循环执行器。它负责回答四个问题下一步该调用模型还是调用工具调用模型时上下文里应该放什么调用工具后结果如何反馈给模型什么时候循环停止2.3 四层架构模型从 DeepSeek-Honeycomb 这类项目的设计思路看Agent 内核通常可以拆成四层接口/编排层对外暴露 run、stream、stop 等入口负责接收任务、启动循环、返回结果。决策层核心是 AgentLoop负责推理模型的选择、Prompt 模板、工具选择、下一步动作判断。工具层ToolRegistry 管理所有可调用工具包括工具注册、参数校验、执行、结果规范化。记忆层维护短期上下文和长期记忆决定哪些历史消息进 Prompt、哪些进向量库。这四层不是每个 Agent 框架都严格边界分明但理解这个分层能帮你看源码时快速定位“某个功能属于哪个层”。3. DeepSeek-Honeycomb 的模块划分从项目名理解多 Worker 协作3.1 为什么叫 HoneycombHoneycomb 是“蜂巢”的意思。蜂巢的特点是单个蜂室很小但通过大量蜂室的结构化排列整个巢穴能承载巨大的存储和协作能力。这个命名很容易让人联想到 Agent 框架的一种常见内核设计多个 Worker 协作每个 Worker 完成一个子任务最终汇聚成一个完整答案。当然并不是所有叫 Honeycomb 的 Agent 项目都一定采用多 Worker 架构项目名更多是愿景表达。但从 Agent 内核设计的通用趋势看多 Worker 协作确实是当前自研框架的热门方向。3.2 常见模块清单一个类 Honeycomb 架构的自研 Agent 项目src 目录下通常会有这些模块src/ ├── agent/ │ ├── base.py # Agent 基类定义生命周期 │ ├── loop.py # 主循环规划-执行-观察 │ └── worker.py # Worker 抽象子任务执行单元 ├── core/ │ ├── context.py # 上下文管理器 │ ├── memory.py # 记忆存储 │ └── registry.py # 工具注册中心 ├── tools/ │ ├── base.py # 工具基类 │ └── builtin/ # 内置工具 ├── llm/ │ ├── client.py # LLM 客户端封装 │ └── templates.py # Prompt 模板 └── runtime/ ├── executor.py # 执行器协调 Worker └── observer.py # 事件日志/可观测性留意这个结构中的依赖方向agent 层依赖 core 层core 层不依赖 agent 层tools 层只需要实现 core 定义的工具接口。这是内核设计里比较重要的一点_依赖方向要向内层收敛不能让工具反向依赖业务。3.3 三个核心抽象拆开来看内核中有三个抽象是你必须掌握的AgentLoop主循环所有 Agent 行为的骨架。它维护当前状态按照“思考 - 行动 - 观察”的顺序循环执行直到满足退出条件。ToolRegistry工具注册中心一个全局工具列表提供register和execute接口。模型不直接调用 Python 函数而是通过名称和参数描述来触发注册表中的工具。ContextManager上下文管理器决定“这一次调用模型messages 数组里应该传什么”。它负责把系统提示、历史对话、工具结果、用户当前输入组装在一起并处理上下文超长问题。这三个抽象看起来简单但它们之间的协作方式和边界决定了整个 Agent 内核的扩展能力。4. 环境准备与最小骨架设计4.1 环境要求以下示例基于 Python 3.10核心不依赖任何重型 Agent 框架只要有一个可调用的 OpenAI 兼容接口即可。如果你使用的是国内大模型平台只要它提供 OpenAI 兼容的 chat completions 接口就可以套用同样的代码结构。依赖方面最小实现只需要两个库pip install openai版本方面不需要刻意锁死以实际环境安装结果为准。示例的重点是演示内核结构而不是绑定特定 SDK。4.2 目录结构为了便于理解我会把示例项目命名为honeycomb-lab目录结构如下honeycomb-lab/ ├── requirements.txt ├── config.yaml ├── honeycomb/ │ ├── __init__.py │ ├── agent.py │ ├── context.py │ ├── memory.py │ ├── registry.py │ └── loop.py └── examples/ └── demo_agent.py这个结构比真实项目简单很多但保留了内核的四个关键要素Agent 生命周期、Context 管理、记忆存储、工具注册。5. 核心流程拆解规划-执行-观察的循环5.1 为什么是“循环”Agent 之所以能做多步任务是因为它不是“一次问答”而是一个带终止条件的循环。以“查询某城市的天气并提醒是否适合出行”这个任务为例模型发现需要天气数据生成一个工具调用请求。内核执行工具拿到天气数据。工具结果返回给模型。模型基于新信息生成最终回答。如果第 1 步模型判断不需要工具循环会直接进入终止分支模型输出就是最终答案。这个循环最常见的名字是ReActReason Act模式。它的核心思想是模型先推理再行动观察结果后继续推理直到问题解决。5.2 状态机设计主循环可以用一个简单的状态机来管理状态含义下一状态INIT接收任务初始化上下文THINKTHINK调用模型生成下一步动作ACTION 或 FINISHACTION执行工具调用OBSERVEOBSERVE把工具结果写回上下文THINKFINISH输出最终结果结束循环无容易出错的地方是_THINK 状态里模型可能返回三种结果一是普通文本最终答案二是工具调用请求三是既没有文本也没有工具调用空输出。内核必须对这三种情况分别处理尤其是空输出要设置重试或终止机制否则会死循环。5.3 上下文组装时机另一个容易忽视的点是每一次 THINK 状态之前都要重新组装上下文。因为 ACTION 和 OBSERVE 之后消息列表里多了工具调用的请求和结果这些内容必须在下一轮模型调用前被放入 messages。很多新手写 Agent 循环时只在第一轮组装上下文后面直接追加消息导致系统指令被覆盖、历史消息无限膨胀、上下文长度溢出。正确做法是由 ContextManager 统一负责组装逻辑并预留“摘要压缩”的钩子。6. 完整示例实现一个可运行的最小 Agent 内核为了说明原理下面给出一个精简但能运行的 Agent 内核骨架。注意这是教学用最小实现用于理解 Agent 内核的通用设计思路并非 DeepSeek-Honeycomb 项目的原始源码。6.1 工具注册中心registry.py# 文件路径honeycomb-lab/honeycomb/registry.py from typing import Any, Callable, Dict, List, Optional class Tool: 工具描述对象。模型看到的是一份 JSON 描述内核调用的是 fn。 def __init__( self, name: str, description: str, fn: Callable[..., Any], parameters: List[Dict[str, Any]], ): self.name name self.description description self.fn fn self.parameters parameters def to_openai_schema(self) - Dict[str, Any]: 转换为 OpenAI 工具的 JSON Schema 格式。 properties {} required [] for param in self.parameters: pname param[name] properties[pname] { type: param.get(type, string), description: param.get(description, ), } if param.get(required, False): required.append(pname) return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: properties, required: required, }, }, } class ToolRegistry: 工具注册中心管理所有可被 Agent 调用的工具。 def __init__(self): self._tools: Dict[str, Tool] {} def register(self, tool: Tool) - None: self._tools[tool.name] tool def get(self, name: str) - Optional[Tool]: return self._tools.get(name) def list_schemas(self) - List[Dict[str, Any]]: return [tool.to_openai_schema() for tool in self._tools.values()] def execute(self, name: str, arguments: dict) - Any: tool self.get(name) if tool is None: return f错误工具 {name} 不存在请检查工具名称。 try: return tool.fn(**arguments) except Exception as exc: return f工具执行失败{exc}这段代码的核心设计是模型侧看到的是 Schema执行侧触发的才是真实函数。这样隔离的好处是你可以随时替换工具实现而不影响模型对工具的理解。6.2 上下文管理器context.py# 文件路径honeycomb-lab/honeycomb/context.py from typing import Dict, List class ContextManager: 负责组装每次模型调用所需的 messages 数组。 def __init__(self, system_prompt: str, max_history: int 10): self.system_prompt system_prompt self.max_history max_history self.history: List[Dict[str, str]] [] def build_messages(self, user_input: str) - List[Dict[str, str]]: messages [{role: system, content: self.system_prompt}] # 截取最近 max_history 条历史消息避免上下文无限膨胀 messages.extend(self.history[-self.max_history:]) messages.append({role: user, content: user_input}) return messages def add_tool_result(self, tool_call_id: str, tool_name: str, content: str) - None: 把工具执行结果追加到历史里。 self.history.append( { role: tool, tool_call_id: tool_call_id, content: content, } ) def add_assistant_message(self, message: Dict[str, str]) - None: self.history.append(message) def clear(self) - None: self.history.clear()ContextManager 是你以后最先需要扩展的模块。比如历史消息超过上下文窗口时不是直接裁剪而是用 LLM 把旧消息总结成摘要后只保留摘要和最近几条消息。6.3 简单记忆存储memory.py# 文件路径honeycomb-lab/honeycomb/memory.py from typing import Dict, List, Optional class MemoryStore: 极简的内存级记忆存储。生产环境可替换为 Redis 或向量库。 def __init__(self): self._store: Dict[str, List[dict]] {} def add(self, session_id: str, entry: dict) - None: self._store.setdefault(session_id, []).append(entry) def get(self, session_id: str) - List[dict]: return self._store.get(session_id, []) def search(self, session_id: str, keyword: str, limit: int 3) - List[dict]: 最简单的关键词召回示例。真实项目通常是向量检索。 all_entries self.get(session_id) matched [e for e in all_entries if keyword in e.get(content, )] return matched[-limit:]这个 MemoryStore 只能算一个占位实现。真实项目里长期记忆往往需要持久化到数据库并通过向量相似度做召回。但它的接口设计可以保留add 写get 读search 召回后面无论底层换成 Elasticsearch 还是向量库接口都不需要大改。6.4 主循环loop.py# 文件路径honeycomb-lab/honeycomb/loop.py import json from typing import Any, Dict, Optional from openai import OpenAI from .context import ContextManager from .registry import ToolRegistry class AgentLoop: Agent 主循环THINK - ACTION - OBSERVE - FINISH。 def __init__( self, client: OpenAI, model: str, context_manager: ContextManager, tool_registry: ToolRegistry, max_steps: int 5, ): self.client client self.model model self.context context_manager self.tools tool_registry self.max_steps max_steps def run(self, user_input: str) - str: messages self.context.build_messages(user_input) step 0 while step self.max_steps: step 1 response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.tools.list_schemas(), tool_choiceauto, ) message response.choices[0].message # 情况 1模型没有请求调用工具直接返回最终答案 if not message.tool_calls: self.context.add_assistant_message( {role: assistant, content: message.content} ) return message.content # 情况 2模型请求调用工具先记录助手消息再执行工具 self.context.add_assistant_message( { role: assistant, content: message.content or , tool_calls: [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in message.tool_calls ], } ) for tc in message.tool_calls: tool_name tc.function.name try: arguments json.loads(tc.function.arguments) except json.JSONDecodeError: arguments {} result self.tools.execute(tool_name, arguments) # 把工具结果追加到 messages同时写入 ContextManager messages.append( { role: tool, tool_call_id: tc.id, content: str(result), } ) self.context.add_tool_result(tc.id, tool_name, str(result)) # 循环继续重新调用模型带上工具观察结果 messages self.context.build_messages(user_input) return 已达到最大步数任务提前终止。建议拆分子任务或检查 Prompt。这段代码是 Agent 内核的心脏它展示了前面反复强调的循环关系模型决策、工具执行、结果回填、再次调用模型。实际项目中这个循环至少还需要加三样东西流式输出、异常重试、步骤级日志。6.5 把整个内核组装起来demo_agent.py# 文件路径honeycomb-lab/examples/demo_agent.py from openai import OpenAI from honeycomb.context import ContextManager from honeycomb.loop import AgentLoop from honeycomb.registry import Tool, ToolRegistry def get_weather(city: str) - str: 模拟天气查询。 weather_map { 北京: 晴气温 25°C空气质量良好, 上海: 多云气温 27°C有微风, 广州: 雷阵雨气温 30°C注意带伞, } return weather_map.get(city, f暂不支持城市{city}) def main() - None: client OpenAI() # 这里会读取环境变量中的 API Key 和 Base URL registry ToolRegistry() registry.register( Tool( nameget_weather, description查询指定城市的实时天气情况, fnget_weather, parameters[ { name: city, type: string, description: 城市名称例如北京, required: True, } ], ) ) ctx ContextManager( system_prompt你是一个有帮助的助手可以调用工具获取信息。, max_history10, ) agent AgentLoop( clientclient, modelgpt-4o-mini, context_managerctx, tool_registryregistry, max_steps5, ) result agent.run(北京天气怎么样适合跑步吗) print(result) if __name__ __main__: main()运行这个示例前需要配置好 OpenAI 兼容接口的环境变量export OPENAI_API_KEY你的_API_Key # 如果使用第三方 OpenAI 兼容接口再设置 # export OPENAI_BASE_URLhttps://你的接口地址/v1然后执行cd honeycomb-lab python examples/demo_agent.py7. 运行验证与效果检查7.1 预期效果如果一切正常Agent 会先调用get_weather工具拿到北京的天气结果然后基于工具返回值生成最终回答。输出大概是北京今天天气晴朗气温 25°C空气质量良好非常适合户外跑步。关键判断标准是输出里的天气数据不是模型凭空生成的而是来自工具的真实返回值。你可以在get_weather函数里故意造一个异常数据比如把北京的温度改成 25°C然后看模型是否复述这个数字。如果复述说明工具调用链路已经通了。7.2 如何验证循环执行过程为了确认循环确实“执行了多轮”可以在loop.py的 while 循环里加一行临时日志print(f[step {step}] 工具调用: {tool_name}, 参数: {arguments}, 结果: {result})运行后你会在终端看到类似这样的输出[step 1] 工具调用: get_weather, 参数: {city: 北京}, 结果: 北京晴气温 25°C空气质量良好这一步如果成功说明 THINK - ACTION - OBSERVE - THINK 的闭环已经生效。你看到的不是一次模型输出而是一个完整的决策循环。7.3 观测自定义工具你也可以注册一个自定义工具来验证扩展性。比如加一个计算器工具registry.register( Tool( namecalculator, description计算两个数字的加减乘除, fnlambda a, b: str(a b), parameters[ {name: a, type: number, description: 第一个数字, required: True}, {name: b, type: number, description: 第二个数字, required: True}, ], ) )然后问 Agent“123 加 456 等于多少”如果模型选择调用 calculator并返回 579说明工具注册、Schema 描述、参数传递这条链路都是通的。8. 常见问题与排查思路问题现象可能原因排查方式解决方案Agent 一直循环不返回最终结果模型每次都请求调用工具但工具结果没有让模型满足退出条件打印每一步的 messages观察模型是否能看到 tool 结果检查工具结果是否成功写入 messages适当增加 max_steps优化 system prompt 明确“信息足够后直接回答”模型生成了 JSON 但解析失败模型返回的 tool_calls 参数不是合法 JSON在 json.loads 前打印 arguments 原文增加容错解析使用更强模型提示模型“严格按 JSON 格式输出”工具执行了但模型不基于结果回答工具结果没有正确回传给模型确认 messages 中 roletool 的消息是否带正确 tool_call_id检查 ContextManager 里 add_tool_result 的实现确保 tool_call_id 一致上下文越来越长最终超出模型限制没有做上下文裁剪或摘要压缩查看 messages 中历史消息数量实现 max_history 截断引入摘要压缩把长期记忆转到向量库工具本身报错导致循环中断工具函数抛异常且未被捕获查看终端堆栈在 ToolRegistry.execute 里捕获异常并返回错误信息让模型知道“工具调用失败”而不是中断进程Agent 调用了不存在的工具工具 Schema 列表与代码注册不一致打印 registry.list_schemas() 对比统一从注册中心生成 Schema禁止手动维护两份列表9. 最佳实践与工程建议9.1 工具层要遵循最小权限原则Agent 内核的设计里最危险的不是决策循环而是工具权限过大。如果一个 Agent 同时注册了“删除数据库”“发送邮件”“读取文件”等工具而内核没有权限校验层Prompt 注入就可能导致灾难性操作。建议在 ToolRegistry 外层增加一个授权层按工具名和参数做白名单校验至少在生产环境做到“默认拒绝、显式放行”。9.2 每一步工具调用都要可观测自研 Agent 内核时日志是最容易被忽略但又最该提前设计的一块。从第一版开始就为每次循环记录结构化日志包括步骤号、模型名称、token 消耗、工具名称、参数、结果摘要、耗时。这样上线后出现“Agent 乱调工具”或者“一直循环”时才能快速定位问题环节。9.3 为工具调用设置超时和幂等工具调用的超时往往比模型调用超时更隐蔽。模型最多几秒到几十秒但工具可能是一个外部 HTTP 请求遇到上游慢接口会拖死整个循环。建议在 Tool 的封装层支持timeout参数并为可能重复执行的工具设计幂等键。9.4 上下文压缩要尽早实现这里的 ContextManager 只做了简单截断只适合原型验证。一旦 Agent 任务变复杂截断会导致早期关键信息丢失。建议在“max_history 截断”和“摘要压缩”之间做分段处理短会话直接保留完整长会话先裁剪“工具返回的大段原文”只保留摘要再长才触发 LLM 摘要。9.5 测试金字塔从工具层开始Agent 应用很难做端到端断言因为 LLM 输出不稳定。但工具层是稳定的应该优先为每个 Tool 写单元测试验证参数校验和返回值格式。然后是 Context 管理器的测试验证不同轮次下 messages 数组的长度和结构。最后才是用固定的 mock 模型响应测试 AgentLoop 的分支逻辑。从下往上测试回归成本会低很多。10. 总结与后续学习方向Agent 内核没有想象中神秘。把一次 LLM API 调用扩展成“规划-执行-观察”的循环再为这个循环配上工具注册中心和上下文管理器一个最小可用的 Agent 内核就已经搭建完成。这篇文章真正想让你带走的不是一个框架的 API 用法而是一种拆解能力拿到任何一个 Agent 开源项目先找它的主循环、找它的工具注册中心、找它的上下文组装逻辑你就能快速判断这个框架的扩展点在哪里、约束在哪里。推荐你按下面的路径继续深入先把文中的最小骨架跑通修改 system prompt、增加新的 Tool、调整 max_steps感受循环的行为变化。然后给内核加一个“记忆持久化”能力把 MemoryStore 换成 Redis 或一个 SQLite 实现。接着研究函数调用的流式输出支持让每一轮思考过程能实时展示给用户。最后挑战更复杂的任务编排比如多 Worker 并行子任务、任务结果的汇聚策略到这一步你就开始触到 Honeycomb 这类“蜂巢”协作式 Agent 架构的设计核心了。如果你在自研 Agent 内核的过程中卡在某一个模块建议先画一张“输入-输出-状态变更”的小图把这一步的输入和输出定义清楚再动手。Agent 内核的复杂度从来不在单个类而在于模块间传递的协议是否清晰。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价