这几天我一直在调一个基于LLM的多步骤任务编排框架，项目代号就叫hermes-agent。取这个名字没有太多花哨的理由，Hermes在神话里是传递消息的信使，而我这套东西干的事情也很类似——把用户的一句自然语言指令，拆解成可执行的小步骤，分发给合适的工具，再把结果收敛成一段人能读懂的回复。整个过程中，它就像一个中间调度层，负责承上启下、串联一切。
如果你正在做AI Agent相关的应用开发，或者你手头有一堆内部API、脚本、数据库查询逻辑，想用自然语言把它们串起来用，那这篇文章应该对你有帮助。我会从设计思路、核心机制、环境搭建、关键代码实现，到常见问题排查，完整拆解一遍。内容偏实操，尽量少讲虚的。
1. 整体设计与思路拆解
1.1 为什么需要这样一个"中间调度层"
先说说我为什么要做hermes-agent，而不是直接调LLM API完事。直接调模型做问答是一回事，但做真正的Agent是另一回事。举个例子，用户说"帮我查一下上周的销售数据，顺便生成一份PDF报告，发到团队邮箱"。如果只靠一次LLM调用，模型顶多给你生成一段查数据库的SQL，或者给你一段Python代码，但不会真的去执行、校验结果、再触发后续动作。
这时候就需要一个编排层，也就是hermes-agent要解决的核心问题：它把LLM从"思考者"变成"调度者"，模型负责理解和规划，真正干活的是一个个注册好的工具。工具可以是函数、API调用、Shell命令、数据库查询，甚至另一个Agent。hermes-agent只做三件事：理解意图、编排步骤、执行并反馈结果。
选择自研而不是直接用现成的Agent框架，主要是因为定制性。很多通用框架把工具调用的协议、消息格式、记忆管理方式都定型了，接入内部系统时往往要做大量适配。而hermes-agent从设计之初就围绕一个原则：简单、透明、可控。它的核心不复杂，复杂的是围绕它扩展的工具生态。
1.2 系统模块是怎么划分的
整个项目可以拆成四个核心模块：调度核心、工具注册中心、记忆管理器和人机交接模块。调度核心负责整个任务的生命周期管理，从接收用户消息开始，到任务完成或需要人工介入为止；工具注册中心维护一份可用工具清单，每个工具包含名称、描述、参数结构、执行函数和权限级别；记忆管理器维护会话上下文和任务中间状态；人机交接模块处理Agent不确定或权限不足的情况。
这四个模块的职责是严格分离的，这一点在后续维护时特别重要。比如我新接入一个内部工单系统，只需要在工具注册中心加一个工具，完全不需要改动调度逻辑。如果我想调整Agent的决策策略，也只动调度核心的文件，不影响工具部分。hermes-agent的设计哲学是"约定大于配置"，每个工具函数只要遵循统一的注册规范，就能被调度核心自动发现和调用。
1.3 技术选型背后的取舍
技术栈上，核心用Python 3.10+，LLM接入层用的是OpenAI兼容接口，异步框架基于asyncio，工具执行放在线程池里。选Python是因为AI生态最成熟，团队内部也最熟悉；但不排斥将来用Go或Rust重写调度核心，因为异步I/O和并发控制在Go里写起来确实很爽。
LLM接入层做成兼容OpenAI接口格式，是考虑到市面上绝大多数模型服务都提供OpenAI兼容的HTTP接口，无论是云端还是私有化部署，这样切换模型的成本几乎为零。我之前试过直接把模型调用写死在业务代码里，后来换模型的时候改到怀疑人生，所以这次坚决把模型交互封装成独立服务，上层只面对统一的chat()接口。
记忆管理器用的是Redis，主要存短期会话上下文；长期记忆用SQLite做持久化。Redis的好处是TTL过期机制很自然，会话超过一定时间自动清理；SQLite则用来存用户偏好、历史任务摘要这类需要跨会话保留的信息。选型没有追求"大而全"，够用、易维护、方便备份就行。
2. 核心流程与关键机制详解
2.1 从一条消息到一次完整行动的链路
先走一遍hermes-agent处理一条用户消息的完整链路，这一步是整个系统的核心，理解了这个，其他的代码都是围绕它展开的。
用户发送消息后，调度核心先做预处理，把当前会话的历史摘要、用户可以调用的工具清单、系统提示词组装好，发送给LLM。LLM的输出不直接返回给用户，而是期望返回一个结构化的JSON，里面包含intent（意图分类）、steps（计划步骤）、requires_clarification（是否需要追问）等字段。调度核心拿到这个JSON后，按步骤依次执行，每执行完一步就把结果回填到上下文里，再决定下一步是继续执行、询问用户还是终止。
这里有一个关键设计：LLM每一轮只做一次规划，而不是一次性把所有步骤都规划完。我在初期版本里试过"全量规划"——让LLM一次性输出一个包含十个步骤的完整计划，然后从头执行到尾。结果是任务执行到第三步时，第四步的前提条件已经不成立了，但计划早就定死了，只能报错。改成"边执行边规划"后，虽然每一轮多了一点延迟，但整体成功率和可解释性都大幅提升。
2.2 工具调用的"翻译层"设计
LLM本身不会调用工具，它只会输出一段文本。所以hermes-agent里有一个专门的"翻译层"，把LLM输出的自然语言步骤转成结构化的工具调用请求。比如LLM输出"我需要调用search_database工具，传入SQL语句SELECT * FROM orders WHERE create_time > '2024-01-01'"，翻译层会把这段文本解析成{"tool": "search_database", "params": {"query": "..."}}，再由执行器去调用。
这个翻译层最初我考虑用正则+关键词匹配，但效果很差，因为LLM的表达方式太灵活了。后来改成用一次额外的LLM小调用，专门做"文本到JSON"的转换，准确率基本能到95%以上。代价是每一轮规划多了一次模型调用，但换来的是极强的兼容性——新增工具时不需要改任何解析逻辑，只要工具描述写清楚，模型自然能学会怎么调用。
工具执行完成之后，返回值也需要过一个"反向翻译层"，把工具返回的一个JSON或表格数据，转成适合LLM理解的文本摘要。这样做的原因是大多数LLM上下文窗口有限，如果把一张一万行的表完整塞回去，很快就把上下文撑爆了。反向翻译层只提取关键统计量、表头信息、异常记录，让LLM有足够的信息做下一步决策，又不至于被海量数据淹没。
2.3 记忆管理：短期与长期分开存
记忆管理是Agent项目里最容易被低估的模块。一开始我图省事，直接把所有历史消息拼在系统提示词里，结果对话超过二十轮之后，token消耗直线上升，模型开始"遗忘"早期的关键信息。后来我加了摘要机制：系统维护一个"滚动摘要"，每五轮对话结束后，让LLM把之前的对话压缩成一段两百字的摘要，替换掉原始历史。
短期记忆继续放在Redis里，key用session:{user_id}，值是最近二十轮的结构化消息。长期记忆则是用户主动声明或系统判断为重要的信息，比如用户偏好、常用查询模板、常去的地点等，这些会写入SQLite。写入长期记忆的动作不是自动的，需要设置一个专门的"记忆写入工具"，LLM在对话中决策是否需要调用它。这样就避免了什么东西都往长期记忆里塞，导致检索时噪声太大的问题。
2.4 人机交接：Agent不是万能的
hermes-agent里有一个非常核心的规则：允许Agent主动说"我不知道"或者"我需要你确认"。很多Agent框架追求全自动，把"需要用户介入"视为失败，但实际操作中，很多任务在关键节点必须有人确认才能继续，比如"确认要给这个客户发送邮件吗？""确认要执行这条删除命令吗？"
我的做法是在工具规范里增加一个requires_confirmation字段。标记了这个字段的工具，在执行前会先停止，向用户展示将要执行的动作和参数，等用户回复确认后才真正调用。同时，规划器有一个内置的置信度阈值，如果LLM给出的步骤置信度低于阈值，系统会主动放弃规划，转为向用户提问澄清，而不是硬着头皮执行。
3. 实操过程与核心环节实现
3.1 环境准备与项目初始化
说了一大堆设计，现在进入实操。先准备环境，我用的是Python 3.10，依赖管理用poetry，主要依赖就四个：openai（LLM接入）、redis（短期记忆）、sqlite3（Python内置，长期记忆）、apscheduler（定时任务，非必需，但做主动提醒时会用到）。
mkdir hermes-agent && cd hermes-agent poetry init # 按提示填写项目信息 poetry add openai redis apscheduler配置文件我会单独建一个config.yaml，避免把密钥写死在代码里。主要配置项包括模型名称、API Base地址、密钥、Redis连接串、工具目录路径等。这里强调一点：API Base一定要可配置，因为你可能用云端模型服务，也可能是公司内网部署的模型网关，写死的话每次切换环境都痛不欲生。
3.2 工具注册中心的实现
工具注册中心是整个系统最基础的组件，它维护一张工具清单，并提供注册、发现、调用三个能力。我用了一段非常简单但很实用的代码来实现工具注册：
# tools/registry.py import inspect import logging from typing import Callable, Dict, Any, Optional logger = logging.getLogger(__name__) TOOL_REGISTRY: Dict[str, Dict[str, Any]] = {} def register_tool( name: str, description: str, parameters: dict, requires_confirmation: bool = False, permission_level: str = "user", ): """装饰器，用于将普通函数注册为Agent可调用的工具。""" def decorator(func: Callable): TOOL_REGISTRY[name] = { "name": name, "description": description, "parameters": parameters, "function": func, "requires_confirmation": requires_confirmation, "permission_level": permission_level, } logger.info(f"[registry] tool registered: {name}") return func return decorator def get_tool_schemas() -> list[dict]: """生成传给LLM的工具描述列表，供模型选择调用。""" schemas = [] for name, meta in TOOL_REGISTRY.items(): schemas.append({ "type": "function", "function": { "name": name, "description": meta["description"], "parameters": meta["parameters"], } }) return schemas def call_tool(name: str, params: dict, user_id: str) -> Any: """执行工具函数，并捕获异常，确保调度核心不受单次工具失败影响。""" if name not in TOOL_REGISTRY: raise ValueError(f"tool not found: {name}") meta = TOOL_REGISTRY[name] if meta["requires_confirmation"]: # 此处返回一个待确认标记，由调度核心处理 return {"__confirmation_required__": True, "tool": name, "params": params} try: result = meta["function"](**params, user_id=user_id) return {"__success__": True, "result": result} except Exception as e: logger.exception(f"[tool:{name}] execution failed: {e}") return {"__success__: False", "error": str(e)}这里的核心设计有几个点值得细说。一是parameters字段严格遵循JSON Schema格式，这样LLM在被问到"这个工具需要哪些参数"时，可以直接通过工具描述里的schema理解，而不需要额外的示例。二是user_id作为隐藏参数自动注入到每个工具函数里，方便做权限控制和数据隔离，新写工具时不需要操心这个参数从哪来，调度器会自动传入。三是工具函数必须返回可JSON序列化的结果，方便后续做上下文回填和日志审计。
3.3 规划器实现：核心调度循环
规划器是整个Agent的"董事会"，它根据用户消息、历史摘要、工具清单，决定下一步做什么。核心是一个循环，每次迭代调用LLM，拿到结构化决策，然后执行动作，把结果反馈给LLM，直到LLM输出"任务完成"。
# core/planner.py import json import asyncio from typing import Optional class Planner: def __init__(self, llm_client, registry, memory_manager, max_iterations=15): self.llm = llm_client self.registry = registry self.memory = memory_manager self.max_iterations = max_iterations async def run(self, user_id: str, user_message: str) -> str: # 获取会话上下文 context = await self.memory.get_context(user_id) tools_schema = self.registry.get_tool_schemas() # 组装系统提示词 sys_prompt = self._build_system_prompt(tools_schema) # 迭代执行 messages = [{"role": "system", "content": sys_prompt}] messages.extend(context["history"]) messages.append({"role": "user", "content": user_message}) for step in range(self.max_iterations): # 请求LLM决策 llm_resp = await self.llm.chat(messages, response_format={"type": "json_object"}) decision = json.loads(llm_resp) if decision.get("status") == "completed": return decision.get("final_answer", "任务完成") if decision.get("status") == "clarification": return decision.get("question", "需要你补充更多信息") # 如果有工具调用 if "tool_calls" in decision: for tc in decision["tool_calls"]: result = self.registry.call_tool( tc["name"], tc["arguments"], user_id ) messages.append({ "role": "assistant", "content": f"调用工具 {tc['name']}，参数：{tc['arguments']}" }) messages.append({ "role": "user", "content": f"工具返回结果：{json.dumps(result, ensure_ascii=False)}" }) else: # 没有工具调用，可能是中间结果或需要继续规划 messages.append({"role": "assistant", "content": llm_resp}) return "执行达到最大迭代次数，任务已停止。"这段代码看起来简单，但有几处细节特别关键。一是response_format强制要求JSON输出，如果没有这个参数，LLM可能会输出一段带解释的文本，解析时就很容易报错。二是在遇到工具调用时，我把"调用工具"这一步作为assistant消息，把工具返回值作为user消息，这样模型能清晰看到"我做了什么"和"世界变成什么样了"。三是必须有最大迭代次数保护，不然Agent可能在某个死循环里出不来，白白消耗token。
3.4 最小可用Agent示例
工具注册、调度循环都写好了，现在把它们串起来做一个最小可用的Agent。我做了三个最基础的示例工具：查询时间、查天气（这里用mock数据）、发个简单的站内信。
# tools/base_tools.py import datetime from tools.registry import register_tool @register_tool( name="get_current_time", description="获取当前的日期和时间，包含星期几。", parameters={ "type": "object", "properties": {}, } ) def get_current_time(user_id: str = None): now = datetime.datetime.now() return {"datetime": now.strftime("%Y-%m-%d %H:%M:%S"), "weekday": now.strftime("%A")} @register_tool( name="send_internal_message", description="向系统内用户发送一条站内消息，接收方通过user_id指定。", parameters={ "type": "object", "properties": { "receiver_id": {"type": "string", "description": "接收方用户ID"}, "content": {"type": "string", "description": "消息内容"} }, "required": ["receiver_id", "content"] }, requires_confirmation=True ) def send_internal_message(receiver_id: str, content: str, user_id: str = None): # 这里接入内部IM系统 return {"status": "sent", "to": receiver_id, "content_preview": content[:20]}这里特别说一下requires_confirmation=True的作用。我故意把"发消息"这个动作标记为需要确认，因为在真实环境里，让Agent未经用户确认就自动给同事发消息是非常危险的，一旦内容有误，造成的尴尬很难挽回。加了确认机制后，调度核心会先向用户展示"即将发送给XXX，内容为YYY，是否确认？"，用户回复确认后才真正执行。这个保护机制成本极低，但价值巨大。
3.5 Agent完整运行实录
工具写好后，我用一个真实场景测试了一下。用户输入："现在几点了？顺便帮我给产品部的王磊发一条站内消息，说我下午三点过去找他开会。"
一次典型的运行过程如下（我开着debug日志，完整记录了下来）：
- 规划器将用户消息、工具schema、历史摘要组装好，发送给LLM。
- 模型返回决策JSON：意图是"查询时间+发送消息"，计划是先后调用
get_current_time，再调用send_internal_message。 - 调度器先调用
get_current_time，拿到当前时间和星期几。 - 调度器尝试调用
send_internal_message，发现该工具标记了requires_confirmation，于是暂停执行，向用户输出："我将向用户王磊（ID: wanglei）发送内容为'我下午三点过去找你开会'的站内消息，请回复确认以继续。" - 用户回复"确认"，调度器继续执行，真正调用
send_internal_message。 - 工具返回发送成功。LLM汇总所有信息，输出："现在是2025年1月15日星期三下午两点零五分。我已向王磊发送站内消息，告知你下午三点过去找他开会。"
整个过程耗时大约八秒，其中真正的工具执行不到零点几秒，大部分时间花在LLM的规划和总结上。这个体验让我觉得还是有优化空间的，比如把"查询时间"这个确定性结果缓存起来，可以再快一点，这个后面再慢慢做。
4. 常见问题与排查技巧实录
4.1 模型返回的JSON反复解析失败
这是我在开发过程中踩过最大的坑。LLM即使被要求输出JSON，偶尔也会在开头加一句解释，或者用Markdown的```json代码块把JSON包起来，导致json.loads直接抛异常。这个问题在长上下文里更容易出现，模型"忘记"了系统提示词里的JSON格式要求。
我的解决方案是写了一个robust_parse_json函数，先尝试直接解析，失败后用正则提取第一个{到最后一个}之间的内容，剔除多余的```标记；如果提取出来的JSON还是缺字段，就触发一次"纠正LLM纠错"——把原始输出和一个明确的错误信息发给模型，要求重新输出合法JSON。这套兜底逻辑把解析成功率从最初的83%提升到了98%以上，剩余的2%基本是模型输出截断，只能靠加大max_tokens或者换更强的模型解决。
4.2 工具调用缺少必需参数
另一个高频问题是模型在调用工具时编造参数。比如我定义了一个工具需要user_id和keyword两个参数，模型可能觉得keyword不重要，就不传。为解决这个问题，我在注册中心给每个参数设了default，但更重要的是在工具schema里把required字段写清楚。实测发现，只要工具描述里把参数说明写得足够细，比如标注"keyword：搜索关键词，必填，不能为空"，模型调用工具时遗漏参数的概率会大幅下降。
还有一种情况是模型传入了schema里没有定义的参数，这通常是因为工具描述里提到了某个概念，模型误以为它是一个参数。比如我在描述里写了"统计订单金额和订单数量"，模型就可能在params里塞一个order_amount字段。针对这个，我在call_tool里做了一层参数白名单过滤，只保留schema里定义过的键，多出来的键直接丢弃并记录warning日志。
4.3 Agent陷入任务循环出不来
试过让Agent查一个很复杂的数据报表，它连续执行了十几次工具调用，每次都在微调SQL语句，但跑出来的都是同样错误的结果。这个问题本质是模型在看不到最终结果的时候，会不断尝试"再调一次碰运气"。我做了两个改进，第一个就是之前的max_iterations保护，默认15轮，超过就强制停止并提示用户；第二个是增加一个"推理摘要"机制，每轮迭代后让模型输出简短的两三句话，说明"为什么这一轮要这样调整"，这一步会显著减少无效尝试，因为模型在输出解释时更容易发现自己逻辑上的漏洞。
4.4 上下文被工具返回结果撑爆
这个问题做数据分析类Agent时一定逃不掉。一次查询可能返回几千行数据，如果全塞进上下文，分分钟打爆token上限。我的方案是给工具返回值加一个"压缩阈值"：规定工具返回的字符串超过800个字符时，由反向翻译层生成一份摘要，只提取行数、列名、前五行样例、关键统计值和异常值标记，原始数据存到临时存储，保留一个result_ref供后续按需获取。这样LLM既能理解结果概况，又不会迷失在细节里，当用户追问"具体哪几行有问题"时，Agent还可以通过另一个工具按引用ID取回原始数据精查。
4.5 权限边界：不该让Agent做的事情一定不能做
最后一个关键提醒，关于Agent的权限控制。如果Agent能调用删除、更新、发送消息这类有副作用的工具，一旦prompt注入或者模型误判，后果可能非常严重。我的建议是三层防护：第一层，工具注册时必须给permission_level字段赋值，user级工具任意调用，admin级工具只有当前用户是管理员时才允许执行；第二层，所有需要确认的工具统一走requires_confirmation机制，宁可多问一次也不能默认执行；第三层，所有工具调用都写入操作日志，包含调用者、工具名、参数、时间、结果，方便事后审计。这三层听起来会让Agent变得不那么"智能"，但我的体会是，在真实业务系统里，安全和可控永远比智能优先。
5. 这个项目后续还能怎么扩展
写完这个基础版本后，我已经在规划几个扩展方向。一个是把Hermes接入更多内部系统，比如工单、监控告警、数据报表平台，目标是让用户通过自然语言就能查工单进度、看系统指标、生成日报。第二个方向是给Agent加主动推送能力，结合定时任务，每天早上自动汇总前一天的销售情况和系统异常，推送到指定的协作群。第三个方向是想把多Agent协作加进来，不是让一个Agent做所有事，而是拆成"管理Agent"和若干个"专业Agent"，管理Agent负责任务分解和结果整合，专业Agent各自负责数据查询、文本生成、代码执行等。这样拆的好处是每个Agent的系统提示词可以更聚焦，不会被各种不相关的指令干扰。
最后再分享一个小技巧。不管是自己写Agent框架，还是用别人开源的框架，一开始一定要尽量保持"薄"。不要一上来就把记忆机制、工具调用、多轮对话、权限体系全做进去，先跑通一个最简单的"LLM+一次工具调用"闭环，再逐步加复杂度。我最早那个版本就是又壮又笨，改一个地方牵一发动全身，重写之后只留最核心的调度逻辑，反而跑得更稳、更好扩展。Agent这个领域变化太快，框架层面的东西越薄，留给未来变化的余地就越大。
