1. 为什么需要给 AI Agent 配一套工程工作流
1.1 从“能聊”到“能干活”的鸿沟
过去一年我花了不少时间在 AI Agent 的落地上,踩过的坑比想象中多得多。最开始我以为只要把大模型的接口接上、把提示词写清楚,Agent 就能自动帮我处理任务。结果现实很骨感:模型确实能“聊”,但一旦让它真正去执行一个多步骤的工程任务,比如读代码、改配置、跑测试、提交结果,它就开始各种翻车——要么忘了上一步做了什么,要么在同一个错误上反复打转,要么把中间结果丢得一干二净。
这个问题的本质,不是模型不够聪明,而是我们缺少一套工程工作流来约束和引导它的行为。所谓 Skills,说白了就是给 AI Agent 配一套“技能包”加“作业规范”,让它知道在什么场景下该调用什么能力、按什么顺序执行、遇到异常怎么回退。这套东西不是提示词的堆砌,而是一套可复用、可组合、可调试的工程化结构。
我后来总结出一个判断标准:如果一个 Agent 只能完成单轮问答,那它就是个玩具;如果它能稳定地完成一个需要五到十个步骤、涉及多个工具调用的任务,那它才算真正进入了工程可用的范畴。而 Skills 就是跨越这条鸿沟的桥。
1.2 Skills 到底解决什么问题
具体来说,Skills 这套工作流主要解决四类问题。
第一类是能力边界模糊。模型不知道自己会什么、不会什么,用户也不清楚该让它干什么。Skills 通过显式定义每个技能的名称、描述、输入输出格式,把能力边界画清楚。比如一个“读取文件”技能和一个“修改文件”技能是分开的,模型不会混着用。
第二类是执行顺序混乱。多步骤任务最怕顺序错乱,Skills 通过工作流编排把步骤固定下来,每一步的产出作为下一步的输入,形成一条清晰的链路。
第三类是错误处理缺失。真实工程任务里失败是常态,Skills 要求每个技能都定义失败时的行为——是重试、是跳过、还是回滚,都要提前想清楚。
第四类是结果不可复现。同样的输入,今天跑出来是这个结果,明天跑出来是另一个结果,这在工程上是不可接受的。Skills 通过固定参数、固定工具版本、固定执行路径,让结果尽可能可复现。
1.3 适合谁来参考这套方案
这套东西不是给纯小白看的,但也不需要你是资深架构师。我的经验是,只要你满足下面任意一条,这套工作流就能帮到你:
- 你已经用过大模型接口,写过一些简单的调用脚本,但想让 Agent 做更复杂的事;
- 你在做自动化工具、RPA、或者内部效率平台,需要把 AI 能力嵌进去;
- 你带团队做 AI 应用,需要一套规范让不同人写的 Agent 能互相配合;
- 你自己折腾过 Agent 框架,但被各种“玄学”问题搞得头大,想要一套更工程化的思路。
下面我会从整体设计、核心细节、实操过程、问题排查四个层面,把这套 Skills 工作流拆开讲清楚。每个部分我都会给出具体的结构、参数和踩坑经验,你可以直接拿去改改用。
2. 整体设计与思路拆解
2.1 为什么选择“技能注册 + 工作流编排”这套结构
市面上做 Agent 的方案大致分两派。一派是“大提示词派”,把所有逻辑塞进一个超长的系统提示里,靠模型自己理解;另一派是“工具调用派”,把能力拆成一个个函数,让模型按需调用。我两种都试过,最后选了后者的变体——技能注册 + 工作流编排。
大提示词派的问题是,提示词一长,模型的注意力就散了。我实测过一个三千字的系统提示,模型在前几轮还能记住关键约束,到第五轮之后就开始忽略一些细节。而且提示词没法做单元测试,改一个字可能影响全局,维护成本极高。
纯工具调用派的问题是,模型有了工具但不知道怎么组合。它可能在一个简单任务上调用五个工具,也可能在复杂任务上只调用一个。缺少编排层,执行路径完全不可控。
技能注册 + 工作流编排的思路是:技能层负责“能做什么”,编排层负责“怎么做”。技能层把每个能力封装成标准接口,编排层用状态机或者有向图把技能串起来。这样模型只需要在编排层做决策,技能层保持稳定。两层分离之后,调试和替换都方便很多。
2.2 技能注册表的设计考量
技能注册表是整个工作流的基础。我设计的注册表包含以下字段:
| 字段名 | 类型 | 说明 |
|---|---|---|
| skill_id | string | 技能唯一标识,用下划线命名 |
| name | string | 人类可读的技能名称 |
| description | string | 一句话说明技能用途,给模型看 |
| input_schema | object | 输入参数的结构定义 |
| output_schema | object | 输出结果的结构定义 |
| handler | function | 实际执行的函数 |
| retry_policy | object | 重试策略,含次数和间隔 |
| timeout | number | 超时时间,单位秒 |
| tags | array | 标签,用于分类和检索 |
这里有几个设计决策值得展开说。
为什么用 schema 而不是自由文本描述输入输出?因为模型对结构化信息的理解远好于自然语言。我试过用自然语言描述“这个技能需要一个文件路径和一个可选的编码格式”,模型经常漏掉可选参数或者把参数名写错。换成 JSON Schema 之后,参数错误率下降了大概七成。
为什么要有 tags?当技能数量超过二十个之后,模型在选择技能时开始出现“选择困难”。tags 让编排层可以先按标签过滤,再在候选集里做选择。比如一个任务涉及“文件操作”,编排层就只把 tags 含 file 的技能给模型看,候选集从二十个降到五个,准确率明显提升。
retry_policy 为什么放在技能层而不是编排层?因为不同技能的失败特性不一样。读文件失败可能是路径问题,重试没用;调接口失败可能是网络抖动,重试有用。把重试策略放在技能层,每个技能可以有自己的默认策略,编排层只在特殊情况下覆盖。
2.3 工作流编排的三种模式
编排层我实现了三种模式,分别应对不同复杂度的任务。
线性模式最简单,技能按固定顺序执行,上一步的输出直接作为下一步的输入。适合流程固定的任务,比如“读取配置 → 校验配置 → 应用配置”。
分支模式在某个节点根据条件选择不同路径。比如“检查文件是否存在”,存在就走“读取”,不存在就走“创建”。分支条件可以是上一步的输出,也可以是外部状态。
循环模式用于需要反复执行直到满足条件的场景。比如“修改代码 → 跑测试 → 如果失败则根据错误信息再修改”,这个循环最多跑 N 次,超过就报错退出。
这三种模式可以嵌套组合。我实际项目里最复杂的任务用了两层嵌套:外层是循环,内层是分支加线性。关键是每一层都要有明确的退出条件,否则容易死循环。
2.4 状态管理:为什么不能只靠对话历史
很多人做 Agent 喜欢把状态放在对话历史里,靠模型自己记住。我一开始也这么干,后来发现两个致命问题。
一是上下文长度限制。任务步骤一多,历史就爆了,模型开始遗忘早期信息。二是状态污染。对话历史里混着用户的闲聊、模型的思考、工具的返回,模型很难从中准确提取当前状态。
我的做法是引入一个独立的状态存储,用一个 JSON 对象保存当前工作流的所有关键信息:当前步骤、已完成步骤、每步的输入输出、全局变量、错误记录。这个状态对象在每一步开始时注入到模型的上下文里,模型只看到它需要的那部分,而不是全部历史。
这样做的好处是状态可序列化、可恢复、可调试。任务跑到一半失败了,我可以把状态存下来,修完 bug 从断点继续,不用从头再跑。
3. 核心细节解析与实操要点
3.1 技能定义的五个关键要素
写一个技能看起来简单,但要写得让模型好用、让自己好维护,有五个要素必须认真对待。
第一是命名。技能名要用动词开头,比如read_file、validate_config、send_notification。不要用file_reader这种名词形式,因为模型在决策时想的是“我要做什么动作”,动词命名更符合它的思考方式。我做过对比测试,动词命名的技能被正确调用的概率比名词命名高约两成。
第二是描述。描述要回答三个问题:这个技能做什么、什么时候用、有什么限制。比如read_file的描述可以写成“读取指定路径的文件内容。适用于需要获取文件原始内容的场景。不支持二进制文件,单文件大小限制 10MB。”这样模型在决策时就有充分依据。
第三是参数设计。参数宁少勿多,必填参数控制在三个以内。可选参数要有合理默认值。参数名要自解释,不要用p1、arg这种。如果某个参数有枚举值,一定要在 schema 里列出来,模型对枚举的遵循度很高。
第四是返回值。返回值要结构化,包含success、data、error三个字段。success是布尔值,data是成功时的结果,error是失败时的错误信息。这样编排层可以统一处理,不用为每个技能写不同的判断逻辑。
第五是幂等性。尽量让技能幂等,也就是同样的输入执行多次结果一样。这对重试很重要。如果一个技能不幂等,重试就可能产生副作用。比如“发送通知”不幂等,重试就会发多条。这种情况要么加去重逻辑,要么在重试策略里明确禁止重试。
3.2 编排层的决策逻辑怎么写
编排层的核心是“下一步做什么”的决策。我的实现是把当前状态和可用技能列表一起给模型,让模型输出下一步要调用的技能和参数。
这里有个关键技巧:不要问模型“下一步做什么”,而是问它“从这些选项里选一个”。开放式提问会让模型发散,给出各种奇怪的方案;封闭式选择则把它的输出限制在可控范围内。
具体做法是,编排层先根据当前状态和任务类型,筛选出一批候选技能,然后把候选技能的名称和描述列给模型,让它选。如果模型选的技能不在候选集里,就报错并要求重选。这样既保留了模型的灵活性,又保证了执行路径的可控性。
另一个技巧是给模型提供少量示例。在编排提示里放两三个“状态 → 选择”的示例,模型的决策准确率会明显提升。示例要覆盖典型场景,包括正常流程和异常分支。
3.3 错误处理的分层策略
错误处理我分了三层。
技能层处理技能内部的错误。比如文件不存在、网络超时、参数格式错误。技能层负责捕获这些错误,包装成标准错误格式返回,并根据 retry_policy 决定是否重试。
编排层处理技能返回的错误。如果某个技能最终失败了,编排层要决定是换一个技能、跳过这一步、还是终止整个工作流。这个决策也交给模型,但要在提示里明确各种错误对应的处理建议。
工作流层处理整体失败。如果工作流终止了,要保存当前状态,记录失败原因,并通知相关人员。这一层是兜底的,确保任何失败都有记录、可追溯。
三层各司其职,不要混在一起。我见过有人把所有错误处理都塞在编排层,结果编排逻辑变得极其复杂,改一处崩一片。
3.4 参数传递与数据转换
技能之间的参数传递是个容易被忽视的细节。上一步的输出往往不能直接作为下一步的输入,需要做转换。
比如read_file返回的是文件内容字符串,而parse_json需要的是一个 JSON 对象。中间就需要一个转换步骤。我的做法是在编排层加一个轻量的数据映射机制,用类似{{step1.output.data}}的模板语法引用上一步的结果,并支持简单的转换函数如json_parse、trim、split。
这个映射机制要简单,不要搞成图灵完备的表达式语言,否则调试成本太高。我限制只支持属性访问和几个内置函数,够用就行。
注意:数据映射的失败要单独处理。如果模板引用的字段不存在,或者转换函数执行失败,要给出明确的错误信息,指出是哪个步骤的哪个字段出了问题。含糊的错误信息会让排查变得非常痛苦。
4. 实操过程与核心环节实现
4.1 环境准备与依赖安装
先说一下我的技术栈选择。语言用 Python,因为生态最全,调试也方便。核心依赖就三个:一个大模型客户端库、一个数据校验库、一个日志库。不需要引入重型框架,自己写反而更可控。
pip install openai pydantic loguru这里解释一下为什么选这三个。大模型客户端库负责和模型通信,我选的是官方库,稳定且更新及时。pydantic 用来做 schema 校验,它的类型提示和验证功能非常成熟,能省掉大量手写校验代码。loguru 用来打日志,比标准库的 logging 好用太多,尤其是结构化日志和异常追踪。
环境变量里要配好模型接口的地址和密钥,这些不要写死在代码里。我用一个.env文件管理,配合python-dotenv加载。
4.2 技能注册表的代码实现
先定义技能的数据结构。
from pydantic import BaseModel, Field from typing import Callable, Any, Optional class RetryPolicy(BaseModel): max_retries: int = 2 interval_seconds: float = 1.0 backoff_factor: float = 2.0 class Skill(BaseModel): skill_id: str name: str description: str input_schema: dict output_schema: dict handler: Callable retry_policy: RetryPolicy = Field(default_factory=RetryPolicy) timeout: float = 30.0 tags: list[str] = Field(default_factory=list) class Config: arbitrary_types_allowed = True然后写注册表。
class SkillRegistry: def __init__(self): self._skills: dict[str, Skill] = {} def register(self, skill: Skill): if skill.skill_id in self._skills: raise ValueError(f"技能 {skill.skill_id} 已注册") self._skills[skill.skill_id] = skill def get(self, skill_id: str) -> Optional[Skill]: return self._skills.get(skill_id) def list_by_tags(self, tags: list[str]) -> list[Skill]: return [ s for s in self._skills.values() if any(t in s.tags for t in tags) ] def all_skills(self) -> list[Skill]: return list(self._skills.values())注册表本身很简单,关键是注册的时机。我的做法是在应用启动时统一注册,注册完打印一份技能清单,方便核对。不要在运行时动态注册,那样容易出并发问题。
4.3 一个完整技能的编写示例
以read_file为例,完整写一遍。
import os from pathlib import Path def read_file_handler(path: str, encoding: str = "utf-8") -> dict: try: p = Path(path) if not p.exists(): return {"success": False, "data": None, "error": f"文件不存在: {path}"} if p.stat().st_size > 10 * 1024 * 1024: return {"success": False, "data": None, "error": "文件超过 10MB 限制"} content = p.read_text(encoding=encoding) return {"success": True, "data": content, "error": None} except Exception as e: return {"success": False, "data": None, "error": str(e)} read_file_skill = Skill( skill_id="read_file", name="读取文件", description="读取指定路径的文本文件内容。适用于需要获取文件原始内容的场景。不支持二进制文件,单文件大小限制 10MB。", input_schema={ "type": "object", "properties": { "path": {"type": "string", "description": "文件的绝对路径或相对路径"}, "encoding": {"type": "string", "description": "文件编码,默认 utf-8", "default": "utf-8"} }, "required": ["path"] }, output_schema={ "type": "object", "properties": { "success": {"type": "boolean"}, "data": {"type": ["string", "null"]}, "error": {"type": ["string", "null"]} } }, handler=read_file_handler, retry_policy=RetryPolicy(max_retries=0), timeout=5.0, tags=["file", "read"] )注意这里的 retry_policy 设成了不重试,因为文件不存在这种错误重试也没用。如果是网络相关的技能,就可以设成重试两次。
4.4 工作流引擎的核心循环
工作流引擎的主循环大概是这样的。
class WorkflowEngine: def __init__(self, registry: SkillRegistry, llm_client, max_steps: int = 20): self.registry = registry self.llm = llm_client self.max_steps = max_steps def run(self, task: str, initial_state: dict = None) -> dict: state = { "task": task, "steps": [], "current_step": 0, "variables": initial_state or {}, "status": "running" } while state["current_step"] < self.max_steps: candidates = self._select_candidates(state) decision = self._ask_llm_for_next(state, candidates) if decision["action"] == "finish": state["status"] = "completed" break skill = self.registry.get(decision["skill_id"]) if not skill: state["steps"].append({"error": f"未知技能 {decision['skill_id']}"}) continue result = self._execute_with_retry(skill, decision["params"]) state["steps"].append({ "step": state["current_step"], "skill_id": skill.skill_id, "params": decision["params"], "result": result }) state["current_step"] += 1 if not result["success"] and decision.get("abort_on_failure"): state["status"] = "failed" break return state这个循环里有几个关键点。max_steps是硬性上限,防止死循环。_select_candidates根据当前状态筛选候选技能。_ask_llm_for_next把状态和候选技能给模型,让它决策。_execute_with_retry负责执行和重试。
4.5 提示词模板的设计
编排层的提示词模板我改了十几版,最后稳定下来的结构是这样的。
你是一个工作流编排器。当前任务:{task} 已完成步骤: {steps_summary} 当前可用技能: {skills_description} 请从上述技能中选择下一步要执行的技能,或者如果任务已完成,返回 finish。 输出格式(JSON): { "action": "execute" | "finish", "skill_id": "技能ID", "params": {}, "reasoning": "选择理由", "abort_on_failure": true | false }steps_summary只包含最近五步的摘要,不是全部历史,避免上下文过长。skills_description只包含候选技能,不是全部技能。reasoning字段强制模型给出理由,这个理由在调试时非常有用,能看出模型是怎么想的。
4.6 状态持久化与断点恢复
状态持久化我用的是 JSON 文件,简单可靠。每执行完一步就把状态写一次盘。
import json def save_state(state: dict, path: str): with open(path, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) def load_state(path: str) -> dict: with open(path, "r", encoding="utf-8") as f: return json.load(f)恢复的时候,把状态加载回来,从current_step继续跑就行。这里要注意,如果技能有副作用(比如已经发了通知),恢复时要避免重复执行。我的做法是在状态里记录每步的执行状态,已成功执行的步骤在恢复时跳过。
5. 常见问题与排查技巧实录
5.1 模型选错技能怎么办
这是最常见的问题。模型面对一堆技能,选了一个看起来相关但实际不对的。
排查思路分三步。第一步看候选集是不是太大,如果候选超过十个,模型的选择准确率会明显下降,这时候要加 tags 过滤。第二步看技能描述是不是有歧义,两个技能描述太像,模型就容易混。第三步看提示词里的示例是不是覆盖了当前场景,没有示例的话模型只能靠猜。
我的经验是,技能描述里要明确写出“不适用于什么场景”。比如read_file的描述里写“不支持二进制文件”,模型就不会在需要读图片时选它。
5.2 参数传错或漏传
参数问题通常有三个原因。一是 schema 定义不清晰,模型不知道某个参数是必填的。二是上一步的输出字段名和下一步的输入字段名对不上,映射失败。三是模型理解错了参数含义。
解决办法:schema 里必填参数用required明确标出;字段命名保持一致,比如统一用path而不是有的地方用file_path有的地方用filepath;参数描述里给一个示例值,模型对示例的遵循度很高。
5.3 工作流陷入死循环
死循环的表现是步骤数一直涨但任务没进展。原因通常是循环的退出条件没设计好,或者模型在某个分支上反复横跳。
我的处理办法是加两个保险。一是max_steps硬上限,超过就强制终止。二是进展检测,如果连续三步调用了同一个技能且参数相同,就判定为卡住,强制终止并报错。这两个保险加上之后,死循环基本没再出现过。
5.4 上下文过长导致模型失忆
任务步骤多了之后,上下文会越来越长,模型开始遗忘早期信息。这个问题我在 4.5 节提过,解决办法是只给模型最近几步的摘要,而不是全部历史。另外,关键信息放在状态对象里,每步都重新注入,不依赖模型自己记住。
如果任务确实需要很长的上下文,可以考虑分段执行,把一个大任务拆成几个子任务,每个子任务独立跑,中间结果通过状态传递。
5.5 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 | 解决措施 |
|---|---|---|---|
| 模型选错技能 | 候选集过大或描述歧义 | 检查候选数量和技能描述 | 加 tags 过滤,描述里写明不适用场景 |
| 参数传错 | schema 不清晰或字段名不一致 | 对比 schema 和实际传参 | 明确 required,统一命名,加示例值 |
| 死循环 | 退出条件缺失 | 看步骤日志是否重复 | 加 max_steps 和进展检测 |
| 模型失忆 | 上下文过长 | 看历史长度 | 只给最近摘要,关键信息放状态 |
| 重试产生副作用 | 技能不幂等 | 检查技能是否有副作用 | 加去重逻辑或禁止重试 |
| 状态恢复后重复执行 | 未记录执行状态 | 检查状态里的步骤记录 | 记录每步执行状态,恢复时跳过已完成 |
5.6 几个我踩过的坑
第一个坑是技能粒度过细。我一开始把每个小操作都拆成一个技能,结果技能数量膨胀到五十多个,模型选择困难,编排层也复杂。后来合并了一些技能,控制在二十个以内,效果好很多。技能粒度以“一个完整的业务动作”为准,不要拆到原子操作。
第二个坑是过度依赖模型的推理。我一开始希望模型自己规划整个工作流,结果它规划的路径经常不合理。后来改成编排层给框架、模型做局部决策,稳定性大幅提升。模型擅长局部选择,不擅长全局规划,这个认知很重要。
第三个坑是忽略日志。早期我没认真打日志,出问题只能靠猜。后来加了结构化日志,每步的输入输出、模型决策、执行耗时都记下来,排查效率提升了一个数量级。日志不是可选项,是必选项。
第四个坑是没有做技能的单测。技能本身有 bug,但被工作流的复杂性掩盖了,排查时以为是编排问题,实际是技能问题。后来我给每个技能都写了单测,确保技能本身是可靠的,再往上排查就快多了。
6. 技能组合与扩展思路
6.1 技能的分层组织
技能多了之后要分层。我分成三层:基础技能、组合技能、业务技能。
基础技能是最底层的原子能力,比如读写文件、调接口、解析数据。组合技能是基础技能的组合,比如“读取配置并校验”。业务技能是面向具体业务的,比如“部署服务”。
分层的好处是复用。组合技能可以复用基础技能,业务技能可以复用组合技能。改一个基础技能,上层自动受益。分层也让候选集筛选更精准,业务任务只需要看业务技能和部分组合技能。
6.2 技能的版本管理
技能会迭代,迭代就要有版本。我的做法是在 skill_id 里带版本号,比如read_file_v2。旧版本保留一段时间,新版本上线后观察一段时间再下线旧版本。这样避免升级技能时把正在跑的工作流搞崩。
版本管理还有个好处是 A/B 测试。同一个功能可以有两个版本的技能,编排层根据配置选择用哪个,对比效果。
6.3 从单 Agent 到多 Agent 协作
单个 Agent 的能力有上限,复杂任务可以拆给多个 Agent。每个 Agent 有自己的技能集和工作流,Agent 之间通过消息传递协作。
比如一个“代码审查”任务,可以拆成三个 Agent:一个负责读代码,一个负责分析问题,一个负责生成报告。三个 Agent 各自跑自己的工作流,通过共享状态交换数据。
多 Agent 的难点在协调。我的经验是,Agent 之间的接口要定义清楚,谁负责什么、产出什么格式、什么时候触发下一个,都要明确。不要指望 Agent 之间靠“自然语言”协商,那样太不稳定。
6.4 性能优化的几个方向
工作流跑得慢是常见问题。优化方向有几个。
一是并行执行。没有依赖关系的技能可以并行跑。比如同时读取多个文件,不用串行。编排层要能识别可并行的步骤。
二是缓存。同样的技能同样的参数,结果可以缓存。尤其是那些耗时的、结果稳定的技能,缓存能省大量时间。
三是模型调用优化。编排层的模型调用是瓶颈之一。可以通过减少候选集、简化提示词、用更小的模型来做编排决策来优化。不是所有决策都需要大模型,简单的选择用小模型就够。
四是技能本身的优化。有些技能写得低效,比如每次都重新建立连接。优化技能实现,整体性能也会提升。
7. 我个人的一些实操体会
这套 Skills 工作流我用了大半年,从最初的玩具版本迭代到现在能稳定跑生产任务,中间改了很多。最大的体会是,工程化不是给 AI 加限制,而是给它搭舞台。舞台搭得好,AI 才能发挥出真正的能力。
另一个体会是,不要追求一步到位。我一开始想设计一个完美的工作流引擎,结果卡了很久。后来改成先跑通最简单的线性流程,再逐步加分支、加循环、加错误处理,反而进展快很多。工程的东西都是迭代出来的,不是设计出来的。
还有一点,调试工具比功能更重要。我花了不少时间做状态查看、步骤回放、日志分析这些调试功能,看起来不产出业务价值,但实际省下的排查时间远超投入。做 Agent 系统,可观测性是第一位的。
最后分享一个小技巧:给每个技能加一个dry_run模式,只返回将要执行的操作而不真正执行。调试工作流时先用 dry_run 跑一遍,确认路径正确再真正执行。这个模式帮我避免了很多误操作。
这套东西还在演进,后面我打算加上技能的自动发现、工作流的可视化编辑、以及基于历史数据的路径优化。有进展再分享。