ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

基于LLM的多步骤任务编排:Agent调度层与工具调用设计实战

基于LLM的多步骤任务编排:Agent调度层与工具调用设计实战 这几天我一直在调一个基于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.10LLM接入层用的是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主要依赖就四个openaiLLM接入、redis短期记忆、sqlite3Python内置长期记忆、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(ftool 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_iduser_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_iterations15): 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_asciiFalse)} }) 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( nameget_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( namesend_internal_message, description向系统内用户发送一条站内消息接收方通过user_id指定。, parameters{ type: object, properties: { receiver_id: {type: string, description: 接收方用户ID}, content: {type: string, description: 消息内容} }, required: [receiver_id, content] }, requires_confirmationTrue ) 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_confirmationTrue的作用。我故意把发消息这个动作标记为需要确认因为在真实环境里让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这个领域变化太快框架层面的东西越薄留给未来变化的余地就越大。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进