第1章：Agent是什么（与LLM的区别）
学习本章后，你将：理解Agent与LLM的本质区别，掌握Agent的核心特征，知道什么时候该用Agent而不是LLM。
1.1 LLM是什么（单次推理，无状态）
如果你已经读过《AI实战指南》，你已经知道LLM（大语言模型）的本质——它是一个"超级自动补全器"。
但为了理解Agent，我们需要先精确地定义LLM的能力边界。
LLM的基本工作模式
当你向ChatGPT提问时，背后发生的事很简单：
用户输入 → LLM推理 → 文本输出
一次调用，一次输出，结束。
LLM的核心特征：
| 特征 | 说明 |
|---|---|
| 单次推理 | 一次调用只做一次"预测下一个词"的运算 |
| 无状态 | 每次调用都是独立的，模型不记住"之前发生了什么" |
| 无工具 | 模型无法自己调用外部系统（除非API帮你包装） |
| 无确认 | 模型不会说"我需要确认一下"，它一定会回答 |
| 无规划 | 模型不会说"这个问题分三步解决"，它直接生成最终答案 |
举例： 你问GPT"25乘以37等于多少"，它会直接输出925。但它不是"算出来"的，而是基于训练数据预测的。如果你让它做多步任务——比如"帮我写一个爬虫，先检查robots.txt，再下载页面，最后提取所有链接"——它给你的是一个一次性生成的代码，而不是一步一步执行的过程。
为什么LLM自己不能成为Agent？
LLM本身没有"执行能力"。它像一个只会说话但不会动手的专家：知道怎么做，但不会自己做；不会主动查资料、调API、写文件；不会根据中间结果调整方向；每次回答都是"从零开始"，不积累经验。
这就是为什么我们需要Agent。
1.2 Agent是什么（自主规划、工具调用、循环执行）
一个直觉定义
Agent = LLM + 规划能力 + 工具集 + 记忆 + 执行循环
更直观的说法：Agent是一个"长了手和脚"的LLM。
- LLM 是"大脑"——负责理解、推理、决策
- 工具（Tools/Skills）是"手"——负责执行具体操作（调API、读文件、写数据库）
- 规划器（Planner）是"前额叶"——负责拆解任务、制定步骤
- 记忆（Memory）是"海马体"——负责记住上下文和历史
- 循环（Loop）是"心跳"——不断重复"思考→行动→观察"的过程
Agent的核心特征
特征1：自主规划
Agent接收到一个任务后，不会直接回答，而是先"思考"：
任务："分析公司Q3财报，生成摘要，发送邮件给团队"
Agent的规划：
1. 读取Q3财报PDF文件
2. 提取关键指标（营收、利润、增长率）
3. 生成摘要文本
4. 调用邮件API发送
5. 确认发送结果
Agent自己决定步骤的顺序和依赖关系，而不是等着人类一步步指令。
特征2：工具调用
Agent可以通过调用外部工具来扩展能力：
| 工具类型 | 示例 | 作用 |
|---|---|---|
| 代码执行器 | Python沙箱 | 执行计算、数据处理 |
| 文件系统 | 读/写文件 | 读取文档、保存结果 |
| API调用 | HTTP请求 | 调数据库、调第三方服务 |
| 搜索工具 | Web搜索 | 获取实时信息 |
| Shell命令 | 终端执行 | 系统操作、脚本运行 |
本质区别： LLM只能"说"，Agent能"做"。
特征3：循环执行
Agent不是一次调用就结束的。它在一个循环中执行：
- 思考（Think）：当前状态是什么？下一步做什么？
- 行动（Act）：执行选定的动作（生成文本、调工具）
- 观察（Observe）：获取行动的结果
- 循环（Loop）：回到思考，根据新状态决定下一步
- 终止（Terminate）：任务完成或达到终止条件
1.3 Agent vs LLM：核心区别
| 维度 | LLM | Agent |
|---|---|---|
| 工作方式 | 单次推理，一次输入一次输出 | 循环执行，多步推理+行动 |
| 状态管理 | 无状态（除对话历史拼接外） | 有状态，维护完整的执行上下文 |
| 工具使用 | 不能自主调工具 | 能自主规划和调用工具 |
| 任务类型 | 适合单步任务（翻译、摘要、生成） | 适合多步复杂任务（分析→决策→执行→反馈） |
| 错误处理 | 一次性输出，错了就错了 | 可在循环中修正，从错误中恢复 |
| 记忆 | 仅靠上下文窗口（有限） | 可持久化记忆（数据库/文件） |
| 决策方式 | 直接生成最可能的回答 | 规划→执行→观察→再决策 |
| 输出形式 | 文本/代码 | 文本 + 动作 + 工具调用结果 |
| 适用复杂度 | 低到中 | 中到高 |
| 典型代表 | ChatGPT网页版、Claude对话 | AutoGPT、AIWork Agent、LangChain Agent |
一句话总结：
LLM是"能说会道"的专家，Agent是"能动手执行"的工程师。
1.4 Agent的工作流程
标准Agent循环
一个Agent的完整工作流程可以抽象为以下循环：
┌──────────────┐
│ 用户输入 │
└──────┬───────┘
▼
┌──────────────┐
│ Planner │ ←── 当前状态 + 任务目标
└──────┬───────┘
▼
┌──────────────┐
│ 决策下一步 │
└──────┬───────┘
│
┌────────┴────────┐
▼ ▼
┌────────────┐ ┌────────────┐
│ LLM 推理 │ │ 工具调用 │
│ (思考) │ │ (行动) │
└──────┬─────┘ └──────┬─────┘
│ │
└───────┬────────┘
▼
┌──────────────┐
│ 观察结果 │ ←── 工具返回/LLM输出
└──────┬───────┘
▼
┌──────────────┐
│ 判断是否完成 │
└──────┬───────┘
│
┌───────┴───────┐
▼ ▼
┌──────────┐ ┌──────────┐
│ 继续 │ │ 结束 │ ──→ 返回结果
└─────┬────┘ └──────────┘
│
└── 回到 Planner ──┘
工作流程的五个阶段
| 阶段 | 做什么 | 举例 |
|---|---|---|
| 输入接收 | 接收用户的任务指令 | "分析这个Excel并生成报告" |
| 规划 | 拆解任务为子步骤 | 1.读文件 2.分析数据 3.生成图表 4.写报告 |
| 执行循环 | Think→Act→Observe | 循环直到所有子步骤完成 |
| 终止判断 | 检查任务是否完成 | 所有步骤完成或达到最大循环次数 |
| 结果输出 | 返回最终结果 | 生成的报告文件+摘要 |
代码示例：最小Agent循环
下面是一个简化版的Agent循环实现。它展示了一个Agent的核心骨架：
import json
from openai import OpenAI
class MiniAgent:
"""最小Agent实现——展示Agent循环的核心逻辑"""
def __init__(self, api_key: str, tools: list):
self.client = OpenAI(api_key=api_key)
self.tools = {t["name"]: t["func"] for t in tools}
self.tool_defs = [{"type": "function", "function": t["def"]} for t in tools]
self.messages = [] # Agent的"记忆"
self.max_steps = 10 # 最大循环次数防止无限执行
def run(self, user_input: str) -> str:
"""执行Agent循环"""
self.messages.append({"role": "user", "content": user_input})
for step in range(self.max_steps):
# === Think: LLM决策下一步做什么 ===
response = self.client.chat.completions.create(
model="gpt-4",
messages=self.messages,
tools=self.tool_defs,
tool_choice="auto"
)
msg = response.choices[0].message
self.messages.append(msg)
if not msg.tool_calls:
# 没有工具调用 → Agent认为任务已完成
return msg.content
# === Act: 执行工具调用 ===
for tool_call in msg.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
result = self.tools[tool_name](**arguments)
# === Observe: 将工具返回结果添加到上下文 ===
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
# === Loop: 进入下一轮Think ===
return "达到最大步数，任务终止"
上述不到30行核心逻辑已经让Agent具备：自主规划能力（LLM决定下一步做什么）、工具调用能力（LLM选择工具和参数）、循环执行能力（反复Think→Act→Observe）、记忆能力（messages累积上下文）和终止判断能力（无工具调用=任务完成）。
这就是Agent的本质。你不需要复杂的框架——一个循环 + 一个LLM + 几个工具函数，就是一个Agent。
1.5 什么时候用Agent
不是所有任务都需要Agent。很多时候，直接调LLM更简单、更便宜、更可靠。
适合LLM的场景（不需要Agent）
- 单步任务：翻译一段文字、总结一篇文章、改写一段代码
- 知识问答：解释一个概念、回答一个事实性问题
- 内容生成：写邮件、写文案、写简单的脚本
- 聊天对话：客服、闲聊、头脑风暴
这些场景的特征：一次LLM调用就能完成，不需要外部工具，不需要多步操作。
适合Agent的场景（需要Agent）
| 场景类型 | 典型任务 | 为什么需要Agent |
|---|---|---|
| 多步操作 | "分析财报→生成摘要→发送邮件" | 需要多个步骤串行执行 |
| 需要工具 | "查数据库→计算结果→写入报表" | LLM无法直接操作数据源 |
| 需要确认 | "检查文件是否存在→如果存在则读取→否则创建" | 需要条件判断和分支逻辑 |
| 需要循环 | "爬取网页→提取链接→继续爬取→汇总结果" | 需要反复执行同类操作 |
| 外部数据 | "搜索最新政策→分析影响→生成报告" | LLM知识有截止日期 |
| 系统集成 | "创建Jira工单→分配负责人→发送Slack通知" | 需要调用企业系统 |
判断框架
收到一个任务时，问自己三个问题：
1. 这个任务需要几步才能完成？
→ 1步 → LLM就够了
→ 多步 → 考虑Agent
2. 这个任务需要调用外部工具吗？
→ 不需要 → LLM就够了
→ 需要 → 考虑Agent
3. 这个任务的中间结果会影响后续步骤吗？
→ 不会 → LLM就够了
→ 会 → 考虑Agent
只要有一个答案是"多步/需要/会"，就应该用Agent。
Agent的价值不在于"更聪明"，而在于"能干活"。
1.6 Agent的局限性
Agent不是万能的。在决定使用Agent之前，必须清楚它的代价和风险。
局限性1：成本大幅增加
每次Agent循环都要调用LLM。一个复杂任务可能涉及5-10次调用，成本是单次LLM的10-100倍。按GPT-4计价，单次约$0.02，一个10步Agent任务可能达到$0.5-1.5。如果你只需要简单翻译，用Agent就是杀鸡用牛刀。
局限性2：延迟累积
LLM单次推理已有1-3秒延迟。Agent的多次串行推理意味着3步约3-10秒，10步约10-30秒，50步可能1-3分钟。用户不一定有耐心等。 设计Agent产品时，需用流式输出、进度反馈、异步执行来缓解等待问题。
局限性3：错误累积与级联失败
Agent的错误不是独立的——上一步的错误会污染后续所有步骤。更糟糕的是：Agent可能陷入死循环、使用错误的工具、或被工具的返回结果误导。
解决方案： 设置最大循环次数、加入验证步骤、人工审核关键节点。
局限性4：Hallucination放大
LLM本身就会"胡说八道"。Agent让这个问题更严重：它可能幻觉出不存在的工具参数、编造工具调用的结果、甚至声称任务已完成而实际上什么都没做。某Agent被要求"查询用户订单状态"，没有真正调用订单API，而是自己编造了一个状态返回——因为它"觉得"用户应该是这样。
局限性5：调试困难
LLM调用是非确定性的，Agent决策路径不可预测，错误定位困难——是LLM推理错了、工具返回错了、还是Agent逻辑错了？传统调试方法几乎失效。 需要专门工具（如LangSmith、AIWork Trace）来追踪执行链。
什么时候不要用Agent
| 场景 | 原因 | 替代方案 |
|---|---|---|
| 简单问答 | 成本高、延迟高 | 直接LLM |
| 需要确定性输出 | Agent输出不稳定 | 传统代码+规则引擎 |
| 实时要求高 | Agent延迟不可控 | 直接LLM或手工编写逻辑 |
| 预算有限 | Agent成本线性增长 | 限制Agent复杂度 |
| 团队没有LLM经验 | 调试困难，运维复杂 | 先学会用LLM，再引入Agent |
1.7 本章小结
| 要点 | 内容 |
|---|---|
| LLM的本质 | 单次推理、无状态、无工具、一次输出 |
| Agent的本质 | LLM + 规划 + 工具 + 记忆 + 循环 |
| 核心区别 | LLM是"知道怎么做"，Agent是"真的能做" |
| 工作流程 | Think → Act → Observe → Loop → Terminate |
| 适用场景 | 多步操作、需要工具、条件分支、系统集成 |
| 主要局限 | 成本高、延迟大、错误累积、调试困难 |
| 关键原则 | 按需选择：简单任务用LLM，复杂任务用Agent |
1.8 思考与练习
-
反思：你当前的工作中，哪些任务适合用LLM直接解决？哪些任务值得用Agent来提效？
-
判断练习：以下任务适合用LLM还是Agent？
- a. "将以下中文翻译成英文"
- b. "读取MySQL数据库中的用户表，计算每个城市的用户数，生成JSON报告"
- c. "写一首关于春天的五言绝句"
- d. "监控服务器日志，当错误率超过5%时发送报警短信"
-
动手实践：打开你常用的AI工具（ChatGPT、Claude等），尝试给一个需要多步操作的任务（例如"搜索今天的科技新闻，选3条最有趣的，写一个摘要，保存到本地文件"），观察它是否能完成——你会发现，纯LLM无法完成需要外部工具的任务。
-
思考题：Agent的"记忆"为什么重要？如果一个Agent没有记忆能力（每次循环都从零开始），会出现什么问题？
下一章：第2章：AIWork底座架构解析——我们将深入一个真实可用的Agent系统，理解它的整体架构和核心组件。
第2章：AIWork底座架构解析
学习本章后，你将：理解AIWork的核心架构，掌握底座组件的职责，能够独立部署和配置AIWork环境。
2.1 AIWork是什么
AIWork不是又一个AI框架。它是一个企业级Agent底座——面向生产环境的、可扩展的、与LLM解耦的Agent运行平台。
2.1.1 定位
如果你熟悉云原生生态，可以把AIWork理解为"Agent界的Kubernetes"：
- Kubernetes管理容器，AIWork管理Agent
- Kubernetes提供Pod、Service、Deployment等抽象，AIWork提供Agent、Skill、Tool、Pipeline等抽象
- Kubernetes不关心你跑的是什么应用，AIWork不关心你用的是哪个LLM
AIWork的核心定位：
| 维度 | 说明 |
|---|---|
| 不是 | 一个具体的AI应用（如聊天机器人、代码助手） |
| 不是 | 一个LLM封装库（如LangChain、LlamaIndex） |
| 是 | Agent的运行平台和生命周期管理平台 |
| 是 | 企业级Agent基础设施的标准底座 |
2.1.2 设计目标
AIWork从设计之初就面向以下五个目标：
目标1：与LLM解耦
AIWork不绑定任何大模型。它通过统一的Adapter层支持OpenAI、Claude、通义千问、GLM、文心一言等主流模型，且支持在运行时动态切换。
目标2：可插拔的Tool体系
Agent的能力来自Tool，而非硬编码。Tool可以是一个Python函数、一个Go服务、一个Shell脚本、甚至是一个外部API。Tool的注册、发现、调用全部标准化。
目标3：企业级稳定性
生产环境的Agent不能"试一下再说"。AIWork提供完备的任务队列、超时控制、重试策略、熔断降级、审计日志——这些都是企业上线必须的能力。
目标4：Skill驱动
Skill是AIWork最核心的抽象。一个Skill = 一组Tool + 编排逻辑 + Prompt模板。Skill可以被独立开发、测试、版本管理，也可以被多个Agent共享。
目标5：渐进式部署
从单机开发调试，到三节点集群，再到多数据中心高可用——同一套架构支持全生命周期演进。
2.2 核心架构图解
AIWork采用六层架构，自顶向下分别为：API Gateway、Task Dispatcher、Agent Runtime、Tool Registry、Skill Engine、LLM Adapter。
┌─────────────────────────────────────────────────────────────────┐
│ 用户 / 外部系统 │
│ (HTTP/gRPC/WebSocket/消息队列) │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 认证 │ │ 限流 │ │ 路由 │ │ 审计日志 │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
│ API Gateway 层 (接入层) │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 任务队列 │ │ 优先级 │ │ 超时控制 │ │ 重试 │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
│ Task Dispatcher 层 (调度层) │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 执行环境 │ │ 状态管理 │ │ 沙箱 │ │ 生命周期 │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
│ Agent Runtime 层 (运行层) │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 工具注册 │ │ 版本管理 │ │ 参数校验 │ │ 健康检查 │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
│ Tool Registry 层 (工具层) │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 技能解析 │ │ 编排引擎 │ │ Prompt │ │ 输出校验 │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
│ Skill Engine 层 (技能层) │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ OpenAI │ │ Claude │ │ 通义千问 │ │ GLM │ │
│ │ Adapter │ │ Adapter │ │ Adapter │ │ Adapter │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
│ LLM Adapter 层 (模型层) │
└─────────────────────────────────────────────────────────────────┘
各层职责一句话总结：
| 层 | 一句话 |
|---|---|
| API Gateway | 谁来、能做什么、每秒能做几次 |
| Task Dispatcher | 任务排队、谁先做、超时了怎么办 |
| Agent Runtime | Agent怎么跑、状态在哪、是否安全 |
| Tool Registry | 工具有哪些、什么版本、怎么用 |
| Skill Engine | 技能怎么拆解、步骤怎么编排、Prompt怎么拼 |
| LLM Adapter | 用哪个模型、调不通怎么办 |
2.3 关键组件详解
2.3.1 API Gateway
API Gateway是AIWork的唯一入口。所有外部请求——无论是来自Web UI、CI/CD系统、还是其他微服务——都必须经过API Gateway。
核心职责：
| 功能 | 说明 |
|---|---|
| 认证 | 支持API Key、JWT、OAuth 2.0三种认证方式 |
| 鉴权 | 基于RBAC的细粒度权限控制，可精确到Skill级别 |
| 限流 | Token Bucket算法，支持按用户、按Skill、按全局三级限流 |
| 路由 | 根据请求的Skill名称和版本号，路由到对应的处理链路 |
| 审计 | 记录每个请求的完整调用链，用于事后追溯 |
限流配置示例（Go）：
// 限流器配置结构体
type RateLimiterConfig struct {
GlobalRate int // 全局每秒请求数
UserRate int // 单用户每秒请求数
SkillRate int // 单Skill每秒请求数
BurstSize int // 突发流量上限
WindowDuration time.Duration // 统计窗口
}
// Token Bucket实现
type TokenBucket struct {
rate float64
capacity float64
tokens float64
lastCheck time.Time
mu sync.Mutex
}
func (tb *TokenBucket) Allow() bool {
tb.mu.Lock()
defer tb.mu.Unlock()
now := time.Now()
elapsed := now.Sub(tb.lastCheck).Seconds()
tb.tokens += elapsed * tb.rate
if tb.tokens > tb.capacity {
tb.tokens = tb.capacity
}
if tb.tokens >= 1.0 {
tb.tokens--
tb.lastCheck = now
return true
}
return false
}
2.3.2 Task Dispatcher
Task Dispatcher是AIWork的调度核心。它负责将API Gateway接收到的请求转化为可执行的任务，并调度到合适的Agent Runtime实例。
核心职责：
| 功能 | 说明 |
|---|---|
| 任务队列 | 基于内存或Redis的队列，支持FIFO和优先级两种模式 |
| 优先级调度 | 支持3级优先级（高/中/低），高优先级任务可抢占低优先级资源 |
| 超时控制 | 每个任务可设定超时时间，超时后自动取消并返回错误 |
| 重试策略 | 支持指数退避重试，可配置最大重试次数和退避系数 |
| 负载均衡 | 将任务分发到负载最低的Agent Runtime节点 |
任务调度核心逻辑（Go）：
// 任务定义
type Task struct {
ID string
SkillName string
SkillVersion string
Input map[string]interface{}
Priority int // 0=低, 1=中, 2=高
Timeout time.Duration
MaxRetries int
CreatedAt time.Time
RetryCount int
}
// 调度器核心循环
func (d *Dispatcher) Start(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
default:
// 按优先级从高到低取出任务
task := d.queue.Dequeue()
if task == nil {
time.Sleep(100 * time.Millisecond)
continue
}
// 创建带超时的Context
taskCtx, cancel := context.WithTimeout(ctx, task.Timeout)
// 异步执行
go func(t *Task, cancelFn func()) {
defer cancelFn()
result, err := d.executeTask(taskCtx, t)
if err != nil {
d.handleFailure(t, err)
return
}
d.handleSuccess(t, result)
}(task, cancel)
}
}
}
2.3.3 Agent Runtime
Agent Runtime是Agent的实际执行环境。每个Agent实例运行在一个隔离的Runtime中，拥有独立的上下文和状态。
核心职责：
| 功能 | 说明 |
|---|---|
| 执行环境 | 每个Agent实例运行在独立的goroutine（Go）/进程（Python）中 |
| 状态管理 | 维护Agent的对话历史、中间结果、会话上下文 |
| 沙箱隔离 | 通过Linux Namespace或Docker实现资源隔离和权限控制 |
| 生命周期 | Agent的创建、运行、暂停、恢复、销毁全生命周期管理 |
| 健康检查 | 定期检测Agent实例的健康状态，异常时自动重启 |
2.3.4 Tool Registry
Tool Registry是AIWork的工具市场。所有Agent可用的工具都在这里注册、管理和发现。
核心职责：
| 功能 | 说明 |
|---|---|
| 工具注册 | 支持函数注册、HTTP API注册、gRPC服务注册三种模式 |
| 版本管理 | 每个Tool支持多版本共存，Skill可以锁定使用的Tool版本 |
| 参数校验 | 基于JSON Schema对工具调用参数进行自动校验 |
| 健康检查 | 定期检查外部工具的连通性和响应时间 |
| 调用统计 | 记录每个Tool的调用次数、成功率、平均耗时 |
工具注册示例（Go）：
package main
import (
"context"
"fmt"
"net/url"
"strings"
"github.com/aiwork/sdk-go/registry"
pb "github.com/yourorg/analysis-proto/gen/go"
"google.golang.org/grpc"
)
var reg *registry.ToolRegistry
func init() {
reg = registry.NewToolRegistry()
}
// 方式1：函数注册
func init() {
reg.RegisterFunc(
"git_diff_parser",
"1.0.0",
"解析Git diff输出，提取变更文件列表和变更行数",
map[string]interface{}{
"diff_text": map[string]string{"type": "string", "description": "git diff的原始输出"},
},
[]string{"diff_text"},
parseGitDiff,
)
}
func parseGitDiff(ctx context.Context, diffText string) (map[string]interface{}, error) {
files := []map[string]interface{}{}
lines := strings.Split(diffText, "\n")
for _, line := range lines {
if strings.HasPrefix(line, "+++ b/") || strings.HasPrefix(line, "--- a/") {
continue
}
if strings.HasPrefix(line, "diff --git") {
parts := strings.Split(line, " b/")
if len(parts) >= 2 {
file := map[string]interface{}{"path": parts[1], "change_type": "modified"}
files = append(files, file)
}
}
}
return map[string]interface{}{"files": files, "total": len(files)}, nil
}
// 方式2：HTTP API注册（外部微服务）
func registerHTTPTools() error {
return reg.RegisterHTTP(
"code_review",
"2.0.0",
"http://review-service:8080/api/review",
registry.WithMethod("POST"),
registry.WithHealthCheck("http://review-service:8080/health"),
registry.WithTimeout(30),
)
}
// 方式3：gRPC注册
func registerGRPCTools() error {
conn, err := grpc.Dial("analysis-service:50051", grpc.WithInsecure())
if err != nil {
return fmt.Errorf("dial failed: %w", err)
}
return reg.RegisterGRPC(
"static_analysis",
"1.2.0",
pb.NewStaticAnalysisClient(conn),
registry.WithHealthCheck("analysis-service:50051"),
)
}
2.3.5 Skill Engine
Skill Engine是AIWork的大脑。它接收用户的自然语言请求，解析为可执行的技能流程，并协调各组件完成执行。
核心职责：
| 功能 | 说明 |
|---|---|
| 技能解析 | 将用户的请求匹配到已注册的Skill，提取参数 |
| 编排引擎 | 按Skill定义的步骤顺序执行Tool调用和LLM推理 |
| Prompt管理 | 根据Skill模板和用户输入动态组装Prompt |
| 上下文管理 | 维护多轮对话的上下文，支持长对话的分段压缩 |
| 输出校验 | 对Agent输出做格式校验、内容过滤、脱敏处理 |
Skill定义示例（YAML）：
name: code_review_skill
version: "1.0.0"
description: "自动Code Review，分析PR变更并生成评审意见"
trigger: "review pull request #"
steps:
- id: fetch_diff
tool: git.DiffParser
input:
diff_text: "${input.diff}"
output: parsed_diff
- id: analyze
tool: analysis.StaticAnalyze
version: "1.2.0"
input:
files: "${parsed_diff.files}"
output: analysis_result
- id: llm_review
llm: true
prompt_template: |
你是一个资深的Code Reviewer。
以下是PR的变更信息：
- 变更文件：${parsed_diff.files}
- 静态分析结果：${analysis_result}
请从以下维度进行评审：
1. 代码正确性
2. 性能影响
3. 安全性
4. 可维护性
输出格式要求JSON：
{
"score": "0-100",
"issues": [{"severity": "critical|major|minor", "description": "..."}],
"suggestions": ["..."]
}
output: review_result
- id: format_output
tool: format.Result
input:
review: "${review_result}"
output: final_result
2.3.6 LLM Adapter
LLM Adapter是AIWork的模型接入层。它屏蔽了不同LLM厂商的API差异，提供统一的调用接口。
核心职责：
| 功能 | 说明 |
|---|---|
| 多模型适配 | 支持OpenAI、Claude、通义千问、GLM、文心一言等 |
| 统一接口 | Chat Completion、Embedding、Function Calling三大能力的统一抽象 |
| 自动Fallback | 主模型不可用时自动切换到备用模型 |
| 成本追踪 | 记录每个请求的Token消耗和费用 |
| 响应缓存 | 对幂等请求启用语义缓存，减少重复调用 |
Adapter接口定义（Go）：
// LLM Adapter统一接口
type LLMAdapter interface {
// Chat Completion
Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error)
// Embedding
Embed(ctx context.Context, texts []string) ([][]float64, error)
// 支持的模型列表
Models() []ModelInfo
// 健康检查
Health() error
}
// 请求结构体
type ChatRequest struct {
Model string
Messages []Message
Temperature float32
MaxTokens int
Tools []ToolDefinition
}
// 带Fallback的调用封装
func CallWithFallback(ctx context.Context, adapters []LLMAdapter, req *ChatRequest) (*ChatResponse, error) {
var lastErr error
for i, adapter := range adapters {
resp, err := adapter.Chat(ctx, req)
if err == nil {
return resp, nil
}
lastErr = err
log.Printf("adapter %d failed: %v, try next", i, err)
// 切换前等待
time.Sleep(time.Duration(i+1) * 500 * time.Millisecond)
}
return nil, fmt.Errorf("all adapters failed, last error: %w", lastErr)
}
2.4 数据流转路径
一个完整的请求从发起到返回，经过以下路径：
用户请求
│
▼
┌─────────────────────────────────────────────────────┐
│ 1. API Gateway │
│ ├─ 认证检查（API Key / JWT） │
│ ├─ 限流判断（用户配额是否充足） │
│ ├─ 路由解析（提取 Skill 名称和版本） │
│ └─ 审计日志（记录请求摘要） │
└─────────────────────────┬───────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 2. Task Dispatcher │
│ ├─ 构造 Task 对象（将请求参数封装为 Task） │
│ ├─ 入队（按优先级插入任务队列） │
│ ├─ 调度（选择负载最低的 Runtime 节点） │
│ └─ 返回 TaskID（调用方立即获得 TaskID） │
└─────────────────────────┬───────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 3. Agent Runtime │
│ ├─ 创建 Agent 实例（或从池中获取） │
│ ├─ 加载会话上下文（历史消息 + 状态） │
│ └─ 调用 Skill Engine │
└─────────────────────────┬───────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 4. Skill Engine │
│ ├─ 解析请求（匹配到 Skill） │
│ ├─ 加载 Skill 定义（读取 YAML 编排） │
│ ├─ 按步骤执行： │
│ │ Step 1: 调用 Tool A（通过 Tool Registry） │
│ │ Step 2: 调用 LLM（通过 LLM Adapter） │
│ │ Step 3: 调用 Tool B │
│ │ Step N: 格式化输出 │
│ └─ 返回执行结果 │
└─────────────────────────┬───────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 5. 结果返回 │
│ ├─ Runtime 将结果写入会话状态 │
│ ├─ Dispatcher 更新任务状态为完成 │
│ ├─ Gateway 格式化响应（JSON / SSE / 回调） │
│ └─ 返回给调用方 │
└─────────────────────────────────────────────────────┘
两种调用模式：
| 模式 | 适用场景 | 响应方式 |
|---|---|---|
| 同步 | 短耗时任务（<30秒） | 直接HTTP响应 |
| 异步 | 长耗时任务（>30秒） | 返回TaskID，通过Webhook或轮询获取结果 |
| 流式 | 逐字输出的场景 | SSE（Server-Sent Events） |
2.5 配置体系
AIWork采用单一配置文件 + 环境变量覆盖的配置策略。配置文件默认路径为 /etc/aiwork/config.yaml。
2.5.1 完整配置结构
# AIWork 全局配置
global:
name: "aiwork-prod" # 实例名称
namespace: "default" # 命名空间（多环境隔离）
log_level: "info" # 日志级别：debug/info/warn/error
data_dir: "/data/aiwork" # 数据目录
# API Gateway 配置
gateway:
listen: ":8080" # 监听地址
tls:
enabled: true
cert_file: "/etc/aiwork/cert.pem"
key_file: "/etc/aiwork/key.pem"
auth:
type: "jwt" # 认证类型：api_key / jwt / oauth2
jwt_secret: "${JWT_SECRET}" # 环境变量注入
expire_hours: 24
rate_limit:
global: 1000 # 全局每秒请求上限
per_user: 50 # 单用户每秒请求上限
burst: 200
# Task Dispatcher 配置
dispatcher:
queue_type: "redis" # 队列类型：memory / redis
redis:
addr: "redis:6379"
password: "${REDIS_PASSWORD}"
db: 0
priorities: 3 # 优先级级别数
default_timeout: 60 # 默认超时（秒）
max_retries: 3
retry_backoff: 2.0 # 退避系数
worker_pool_size: 50 # 工作池大小
# Agent Runtime 配置
runtime:
max_instances: 100 # 最大并发Agent实例数
session_ttl: 1800 # 会话存活时间（秒）
sandbox:
enabled: true
type: "docker" # 沙箱类型：docker / namespace / none
memory_limit: "512m"
cpu_limit: 1.0
network: false # 是否允许网络访问
# Tool Registry 配置
tool_registry:
storage:
type: "etcd" # 配置存储：etcd / mysql / file
endpoints: ["etcd:2379"]
health_check_interval: 30 # 健康检查间隔（秒）
version_limit: 5 # 每个工具保留的最大版本数
# Skill Engine 配置
skill_engine:
skill_dir: "/etc/aiwork/skills" # Skill定义文件目录
max_steps: 20 # 单个Skill最大步骤数
max_recursion: 3 # 最大递归深度
prompt_cache_size: 1000 # Prompt模板缓存大小
# LLM Adapter 配置
llm:
default_model: "gpt-4o"
adapters:
- name: "openai"
type: "openai"
api_key: "${OPENAI_API_KEY}"
base_url: "https://api.openai.com/v1"
models:
- "gpt-4o"
- "gpt-4o-mini"
timeout: 60
retry_count: 3
- name: "tongyi"
type: "dashscope"
api_key: "${TONGYI_API_KEY}"
models:
- "qwen-plus"
- "qwen-max"
timeout: 60
- name: "zhipu"
type: "glm"
api_key: "${ZHIPU_API_KEY}"
models:
- "glm-4-plus"
timeout: 60
# 模型Fallback策略
llm_fallback:
enabled: true
primary: "openai/gpt-4o"
fallbacks:
- "tongyi/qwen-max"
- "zhipu/glm-4-plus"
circuit_breaker:
error_threshold: 5 # 连续错误次数触发熔断
recovery_timeout: 30 # 熔断恢复时间（秒）
2.5.2 配置覆盖优先级
默认值 < 配置文件 < 环境变量 < 运行时API
环境变量格式：AIWORK_<SECTION>_<KEY>，例如 AIWORK_GATEWAY_RATE_LIMIT_GLOBAL=2000。
2.6 部署架构
AIWork支持三种部署模式，适应不同阶段的团队需求。
2.6.1 单机部署（开发/测试）
适合个人开发者或小团队试用。所有组件部署在同一台机器上。
┌─────────────────────────────────────┐
│ 单台服务器 │
│ ┌───────────┐ │
│ │ AIWork │ Go 二进制 │
│ │ (全部组件) │ │
│ └─────┬─────┘ │
│ │ │
│ ┌─────┴──────┐ ┌──────────────┐ │
│ │ Redis │ │ etcd/sqlite │ │
│ │ (队列/缓存) │ │ (配置存储) │ │
│ └────────────┘ └──────────────┘ │
└─────────────────────────────────────┘
启动命令：
# 使用sqlite和内存队列（零依赖）
./aiwork server --config /etc/aiwork/config.yaml
# 使用Redis（需要外部Redis）
./aiwork server --config /etc/aiwork/config.yaml \
--dispatcher.queue_type=redis \
--dispatcher.redis.addr=localhost:6379
2.6.2 集群部署（生产环境）
适合正式上线的团队。组件拆分部署，支持水平扩展。
┌─────────────────────┐
│ 负载均衡器 │
│ (Nginx/HAProxy) │
└──────────┬──────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌────┴─────┐ ┌────┴─────┐ ┌────┴─────┐
│ API GW │ │ API GW │ │ API GW │
│ Node 1 │ │ Node 2 │ │ Node 3 │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
┌────┴─────┐ ┌────┴─────┐ ┌────┴─────┐
│ Runtime │ │ Runtime │ │ Runtime │
│ Node 1 │ │ Node 2 │ │ Node 3 │
└──────────┘ └──────────┘ └──────────┘
│ │ │
└──────────────┼──────────────┘
│
┌────────┴────────┐
│ Redis │
│ Cluster │
└─────────────────┘
┌────────┴────────┐
│ etcd │
│ Cluster │
└─────────────────┘
2.6.3 高可用部署（企业级）
适合金融、医疗等对可用性要求极高的场景。多数据中心 + 单元化部署。
# 高可用部署拓扑
regions:
- name: "shanghai"
zone: "sh-b"
replicas: 3
components:
- gateway
- dispatcher
- runtime
storage:
redis_sentinel:
- "redis-sh-1:26379"
- "redis-sh-2:26379"
- "redis-sh-3:26379"
etcd:
endpoints:
- "etcd-sh-1:2379"
- "etcd-sh-2:2379"
- "etcd-sh-3:2379"
- name: "beijing"
zone: "bj-a"
replicas: 3
components:
- gateway
- dispatcher
- runtime
storage:
redis_sentinel:
- "redis-bj-1:26379"
- "redis-bj-2:26379"
- "redis-bj-3:26379"
etcd:
endpoints:
- "etcd-bj-1:2379"
- "etcd-bj-2:2379"
- "etcd-bj-3:2379"
# 跨数据中心同步
cross_region:
sync_mode: "async" # 异步同步，保障本地写入性能
sync_interval: 5 # 同步间隔（秒）
conflict_resolution: "last_write_wins"
2.6.4 部署参数对比
| 维度 | 单机 | 集群 | 高可用 |
|---|---|---|---|
| 节点数 | 1 | 3-9 | 9+ |
| Redis | 单机 | Sentinel/Cluster | 跨机房Cluster |
| 配置存储 | 文件/sqlite | etcd | 多etcd集群 |
| SLA | 无保证 | 99.9% | 99.99% |
| 故障恢复 | 手动 | 自动 | 自动 + 跨机房切换 |
| 适用团队 | 1-3人 | 5-20人 | 20+人 |
2.7 本章小结
| 要点 | 内容 |
|---|---|
| AIWork是什么 | 企业级Agent底座，管理Agent的完整生命周期 |
| 核心架构 | 六层架构：API Gateway → Task Dispatcher → Agent Runtime → Tool Registry → Skill Engine → LLM Adapter |
| 关键组件 | Gateway（接入）、Dispatcher（调度）、Runtime（执行）、Registry（工具）、Engine（技能）、Adapter（模型） |
| 数据流转 | 请求经认证、排队、执行、编排、模型调用五步处理后返回 |
| 配置体系 | 单一config.yaml + 环境变量覆盖，支持多模型适配和Fallback |
| 部署模式 | 单机（开发）→ 集群（生产）→ 高可用（企业级） |
2.8 思考与练习
- 理解：用一句话向团队成员解释AIWork的六层架构中各层的职责。
- 配置：在本地启动一个最小化的AIWork实例（使用内存队列和sqlite），确认它能响应健康检查接口。
- 扩展：设计一个新的LLM Adapter（比如对接DeepSeek或MiniMax），需要实现哪些接口？
- 部署：如果你的团队有3个人，选择一个合适的部署模式，并说明理由。
- 场景：假设你需要让AIWork支持"自动生成API文档"的需求，你会怎么设计对应的Skill和Tool？
下一章：第3章：从0搭一个Agent——我们会动手搭建第一个完整的Agent，让你亲眼看到AIWork跑起来的过程。
第3章：从0搭一个Agent
学习本章后，你将：从0搭建一个可用的AI-Agent，掌握核心组件编写，能够运行和调试Agent。
3.1 准备工作
环境要求
Python >= 3.10
pip install openai python-dotenv
本章提供FakeLLM用于测试，无需真实API密钥。如需连接OpenAI：
# .env
OPENAI_API_KEY=sk-your-key-here
import json, logging, time
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
3.2 定义Agent的输入输出
Agent系统的起点是两个基础类型：Message（通信单元）和Task（任务病历）。
Message
class MessageRole(str, Enum):
SYSTEM = "system"; USER = "user"
ASSISTANT = "assistant"; TOOL = "tool"
@dataclass
class Message:
role: MessageRole
content: str
tool_calls: Optional[list] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
metadata: dict = field(default_factory=dict)
Task
Task记录从创建到完成的完整状态：
@dataclass
class Task:
task_id: str
goal: str
status: str = "pending" # pending→planning→executing→completed/failed
plan: list = field(default_factory=list)
current_step: int = 0
steps_output: list = field(default_factory=list)
result: Any = None
error: Optional[str] = None
def summary(self) -> dict:
return {
"task_id": self.task_id,
"goal": self.goal[:60],
"status": self.status,
"steps_planned": len(self.plan),
"steps_done": self.current_step,
"has_error": self.error is not None,
}
3.3 实现Planner
Planner是Agent的"大脑"——接收用户目标，让LLM生成可执行步骤。
Prompt设计
PLANNER_PROMPT = """You are an AI Agent planner. Given a task and available tools,
create a step-by-step execution plan.
Available tools:
{tools_desc}
Output JSON format:
{{
"steps": [
{{"step": 1, "action": "description", "tool": "tool_name", "input": {{}}}}
],
"reasoning": "why this plan"
}}
"""
Planner类
class Planner:
def __init__(self, llm_client, model: str = "gpt-4"):
self.client = llm_client
self.model = model
def plan(self, goal: str, tools: list) -> list[dict]:
tools_desc = "\n".join(
f"- {t.name}: {t.description}" for t in tools
)
messages = [
{"role": "system", "content": PLANNER_PROMPT.format(tools_desc=tools_desc)},
{"role": "user", "content": f"Task: {goal}"},
]
response = self.client.chat.completions.create(
model=self.model, messages=messages,
response_format={"type": "json_object"}, temperature=0.1,
)
steps = json.loads(response.choices[0].message.content).get("steps", [])
logging.info(f"[Planner] 计划: {len(steps)} 步")
return steps
FakeLLM：免API密钥测试
class FakeLLM:
"""模拟OpenAI客户端，返回预设计划"""
class _R: choices = None
class _C: message = None
class _M: content = None; tool_calls = None
class _Chat:
def create(self, model=None, messages=None, **kwargs):
task = next((m["content"].replace("Task:","").strip()
for m in reversed(messages or []) if m["role"]=="user"), "test")
plan = json.dumps({"steps": [
{"step":1,"action":f"搜索'{task}'","tool":"search_knowledge","input":{"query":task}},
{"step":2,"action":"计算数据","tool":"calculator","input":{"expression":"42*2"}},
{"step":3,"action":"查天气","tool":"get_weather","input":{"city":"北京"}},
],"reasoning":"test"}, ensure_ascii=False)
FakeLLM._M.content = plan
FakeLLM._C.message = FakeLLM._M
FakeLLM._R.choices = [FakeLLM._C]
return FakeLLM._R
def __init__(self):
self.chat = self._Chat()
3.4 实现Executor
Executor是Agent的"双手"——接收每一步，查工具、执行、返回结果。
class Executor:
def __init__(self):
self.execution_log: list[dict] = []
def execute_step(self, step: dict, registry: "ToolRegistry") -> dict:
result = {
"step": step["step"],
"action": step["action"],
"tool": step.get("tool"),
"input": step.get("input", {}),
"output": None, "error": None, "duration_ms": 0,
}
start = time.time()
tool_name = step.get("tool")
if tool_name:
tool = registry.get(tool_name)
if not tool:
result["error"] = f"工具 '{tool_name}' 未注册"
else:
try:
result["output"] = tool.execute(**step.get("input", {}))
except Exception as e:
result["error"] = f"执行异常: {e}"
else:
result["output"] = f"[思考] {step['action']}"
result["duration_ms"] = int((time.time() - start) * 1000)
self.execution_log.append(result)
return result
def get_log(self) -> list[dict]:
return self.execution_log
3.5 实现Tool Registry
Tool Registry是Agent的"工具箱"——统一注册、查找、发现工具。
Tool与Registry
@dataclass
class Tool:
name: str
description: str
fn: callable
parameters: dict # JSON Schema格式
def execute(self, **kwargs) -> Any:
return self.fn(**kwargs)
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
logging.info(f"[Registry] 注册: {tool.name}")
def get(self, name: str) -> Optional[Tool]:
return self._tools.get(name)
def list_tools(self) -> list[Tool]:
return list(self._tools.values())
def to_openai_format(self) -> list[dict]:
return [{"type": "function", "function": {
"name": t.name, "description": t.description,
"parameters": t.parameters,
}} for t in self._tools.values()]
示例工具与注册
def tool_calculator(expression: str) -> str:
allowed = set("0123456789+-*/.()% ")
if not all(c in allowed for c in expression):
return "错误：包含非法字符"
try:
return f"{expression} = {eval(expression, {'__builtins__': {}}, {})}"
except Exception as e:
return f"计算错误: {e}"
def tool_search_knowledge(query: str) -> str:
kb = {"python": "Python是1991年创建的高级编程语言。",
"ai": "AI研究如何让机器模拟人类智能。",
"agent": "AI Agent是能自主规划、执行行动的智能体系统。"}
for k, v in kb.items():
if k in query.lower():
return f"[知识] {v}"
return f"未找到'{query}'的相关信息。"
def tool_get_weather(city: str) -> str:
db = {"北京": "晴，15-25°C", "上海": "多云，20-28°C", "深圳": "阵雨，25-30°C"}
return db.get(city, f"{city}：晴，20-26°C")
TOOL_DEFS = [
Tool("calculator", "数学计算，输入表达式",
tool_calculator, {"type": "object", "properties": {
"expression": {"type": "string"}}, "required": ["expression"]}),
Tool("search_knowledge", "搜索知识库，输入关键词",
tool_search_knowledge, {"type": "object", "properties": {
"query": {"type": "string"}}, "required": ["query"]}),
Tool("get_weather", "查询天气，输入城市名",
tool_get_weather, {"type": "object", "properties": {
"city": {"type": "string"}}, "required": ["city"]}),
]
def setup_default_tools() -> ToolRegistry:
registry = ToolRegistry()
for t in TOOL_DEFS:
registry.register(t)
return registry
3.6 组装运行第一个Agent
Agent主类
class Agent:
def __init__(self, planner: Planner, executor: Executor, registry: ToolRegistry):
self.planner = planner
self.executor = executor
self.registry = registry
self.task_history: list[Task] = []
def run(self, goal: str, dry_run: bool = False) -> Task:
task = Task(task_id=f"task-{int(time.time())}", goal=goal)
self.task_history.append(task)
task.status = "planning"
try:
task.plan = self.planner.plan(goal, self.registry.list_tools())
except Exception as e:
task.status = "failed"; task.error = f"规划失败: {e}"
return task
if dry_run:
task.status = "completed"; return task
task.status = "executing"
for step in task.plan:
task.current_step = step["step"]
sr = self.executor.execute_step(step, self.registry)
task.steps_output.append(sr)
logging.info(f" Step {step['step']} [{'OK' if not sr['error'] else 'FAIL'}] ({sr['duration_ms']}ms)")
task.status = "completed" if not task.error else "failed"
task.result = "\n".join(
f"Step {s['step']} [{'OK' if not s['error'] else 'ERR'}]: {str(s.get('output',''))[:80]}"
for s in task.steps_output)
return task
Agent运行流程图
┌──────────────────────────────────────────────────────────┐
│ Agent 运行循环 │
├──────────────────────────────────────────────────────────┤
│ 用户目标 ──→ ┌──────────┐ ┌──────────────┐ │
│ │ Planner │───→│ 步骤列表 │ │
│ │ (LLM) │ │ Step 1,2...N │ │
│ └──────────┘ └──────┬───────┘ │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ Executor │ │
│ │ ┌──────┐ ┌───────────┐ │ ┌────────────┐ │
│ │ │ 执行 │─→│ Tool/思考 │ │ │ Registry │ │
│ │ └──┬───┘ └─────┬─────┘ │ └────────────┘ │
│ └─────┼─────────────┼────────┘ │
│ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ │
│ │ 结果输出 │ │ 错误处理 │ │
│ └────┬─────┘ └────┬─────┘ │
│ └──────┬───────┘ │
│ ▼ │
│ 步骤未完成 → 下一步 ; 全部完成 → 返回结果 │
└──────────────────────────────────────────────────────────┘
主函数：运行Demo
def setup_logging(level=logging.INFO):
logging.basicConfig(level=level, format="%(asctime)s %(message)s", datefmt="%H:%M:%S")
def run_demo(use_fake=True, api_key=None):
planner = Planner(FakeLLM(), model="fake") if use_fake else Planner(
__import__("openai").OpenAI(api_key=api_key), model="gpt-4")
agent = Agent(planner, Executor(), setup_default_tools())
task = agent.run("了解AI Agent的概念并计算示例数据")
print(f"\n状态: {task.status} | 步骤: {len(task.steps_output)}\n" + "-" * 40)
for s in task.steps_output:
icon, out = ("OK", str(s.get("output",""))[:80]) if not s["error"] else ("FAIL", s["error"])
print(f" [{icon}] Step {s['step']}: {s['action']}\n {out}\n")
print(f"[结果]\n{task.result}")
if __name__ == "__main__":
setup_logging()
run_demo() # FakeLLM模式
# run_demo(False, "sk-...") # 真实LLM模式
3.7 调试技巧
Agent调试比普通程序更复杂——LLM的不确定性让Bug更难复现。
技巧1：结构化日志
setup_logging(logging.DEBUG)
print(json.dumps(task.summary(), indent=2)) # 查看Task快照
技巧2：Dry-Run模式
task = agent.run("我的目标", dry_run=True) # 只规划不执行
print(json.dumps(task.plan, indent=2, ensure_ascii=False))
技巧3：执行日志分析
for e in executor.get_log():
print(f"[{e['duration_ms']}ms] {e['tool']}({e['input']}) -> {str(e['output'])[:50]}")
常见问题
| 问题 | 原因 | 解决 |
|---|---|---|
| Planner返回空计划 | LLM响应格式错误 | 检查response_format，增加few-shot |
| 工具调用失败 | 参数名不匹配 | 检查parameters定义与input一致 |
| 死循环 | 步骤过多 | 添加max_steps限制 |
| 结果不准确 | 预设数据太简单 | 切换到真实LLM或丰富预设 |
本章小结
| 要点 | 内容 |
|---|---|
| Agent架构 | Planner(规划) + Executor(执行) + ToolRegistry(工具管理) |
| Planner | 用LLM将用户目标分解为可执行步骤列表 |
| Executor | 按顺序执行步骤、调用工具、处理异常 |
| ToolRegistry | 统一注册/查找/调用工具，支持OpenAI格式导出 |
| FakeLLM | 无需API密钥即可测试完整Agent流程 |
| 调试方法 | 结构化日志、Dry-Run、执行日志、Task快照 |
完整代码约180行，覆盖Agent核心概念：规划、执行、工具管理、状态追踪。
思考与练习
-
扩展工具：注册一个新工具（如
send_email），修改FakeLLM预设计划，验证流程正常。 -
任务队列：修改Agent支持并发执行多个任务，通过
task_history追踪所有状态。 -
重试机制：在Executor中实现工具调用失败后自动重试1次，记录重试日志。
-
切换真实LLM：用OpenAI API密钥调用
run_demo(use_fake=False, api_key="sk-...")，对比Planner输出差异。 -
思考：如果步骤3依赖步骤2的输出，Task数据结构和Planner应如何设计这种依赖关系？
下一章：第4章：Skill设计与Skill-Engine
第4章：Skill设计与Skill-Engine
学习本章后，你将：掌握Skill的设计原则，能够编写可复用的Skill，理解Skill-Engine的执行机制。
4.1 什么是Skill
在第3章中，你学会了写Prompt。但每个Prompt都是"一次性"的——用完就扔，下次重新写。
如果我们要构建一个真正的AI Agent，不能每次都手写Prompt。
你需要一种机制，把"完成某个任务的能力"封装成一个可复用的单元。这个单元，就是Skill。
Skill的定义
Skill是一个结构化的"能力单元"，它包含：
- 做什么：这个Skill解决什么问题
- 怎么做：用什么Prompt模板、什么参数
- 输入是什么：需要什么数据
- 输出是什么：期望的结果格式
- 怎么执行：怎么调用AI、怎么处理结果
类比：
| 概念 | 传统软件 | AI Agent |
|---|---|---|
| 能力单元 | 函数/方法 | Skill |
| 复用 | 函数调用 | Skill调起 |
| 组合 | 函数A调用函数B | Chain/Orchestration |
| 库 | npm包/PyPI | Skill库 |
一个具体的例子：
{
"name": "translate_text",
"description": "将文本翻译成目标语言",
"input_schema": {
"text": "要翻译的原文",
"target_language": "目标语言，如'英文'、'日文'"
},
"output_schema": {
"translated_text": "翻译后的文本",
"source_language": "检测到的源语言"
},
"prompt_template": "你是一个专业翻译。将以下文本翻译成{target_language}，只返回翻译结果：\n\n{text}"
}
这个translate_text Skill就像是一个"函数"——你传参给它，它返回结果。不同的只是：它的核心逻辑不是代码，而是一个精心设计的Prompt。
Skill vs. 普通Prompt
| 维度 | 普通Prompt | Skill |
|---|---|---|
| 复用性 | 需要手动复制粘贴 | 结构化存储，一键调用 |
| 参数化 | 每次手动替换 | 变量自动注入 |
| 组装 | 不适合 | 可以Chain/组合 |
| 管理 | 散落在各处 | 有分类、版本、命名规范 |
| 测试 | 手动测试 | 可自动化测试 |
Skill的形态
Skill可以有多种存在形式：
- 文本定义：JSON/YAML配置文件，描述Skill的元数据和模板
- 代码函数：在Agent框架中以函数形式注册
- API接口：暴露为远程可调用的微服务
本章我们关注文本定义形态——这是最通用、最容易理解的形式。
4.2 Skill的结构
一个完整的Skill由三部分组成：Metadata（元数据）、Prompt Template（提示模板）、Execution Logic（执行逻辑）。
4.2.1 Metadata（元数据）
元数据是Skill的"身份证"，告诉系统这个Skill是谁、做什么、怎么用。
# metadata
name: code_review # Skill名称，全局唯一
version: 1.2.0 # 语义化版本
author: team-alpha # 作者/维护者
description: > # 功能描述（用于自动匹配）
对代码进行审查，发现潜在问题、安全漏洞和性能瓶颈
category: developer-tools # 分类
tags: [code, review, qa] # 标签（用于搜索）
visibility: public # 可见性：public/private/internal
|
4.2.2 Prompt Template（提示模板）
模板是Skill的"大脑"——它定义了AI怎么处理输入。
# prompt_template
template: |
你是一个资深代码审查专家。你的任务是审查以下代码，重点关注：
1. 潜在Bug（NullPointer、资源泄露、并发问题）
2. 安全漏洞（SQL注入、XSS、越权）
3. 性能问题（不必要的循环、重复查询）
4. 代码规范（命名、格式、设计模式）
--- 待审查代码 ---
语言：{language}
代码：
```{language}
{code_content}
请按以下格式输出：
审查结果
- 严重问题：数量
- 警告：数量
- 建议：数量
问题详情
| 级别 | 类型 | 行号 | 说明 |
总体评价
input_variables:
- name: language type: string description: 代码语言 required: true
- name: code_content type: string description: 完整的代码内容 required: true max_length: 8000
**模板设计要点：**
- 使用`{variable}`语法标记变量位置
- 给AI明确的角色（Role）
- 说明输出格式（方便程序后续处理）
- 提供示例（Few-shot）提升输出质量
---
### 4.2.3 Execution Logic（执行逻辑）
执行逻辑定义了"怎么调用这个Skill"。
```yaml
# execution
engine: chat_completion # 执行引擎类型
model:
provider: openai # 模型提供商
name: gpt-4o # 模型名称
temperature: 0.3 # 温度参数
max_tokens: 4096 # 最大输出长度
retry:
max_attempts: 3 # 失败重试次数
backoff: exponential # 退避策略
timeout: 30000 # 超时时间(ms)
output_parser: structured # 输出解析器
完整Skill定义示例
# skill-definition/code-review.yaml
name: code_review
version: 1.2.0
author: team-alpha
description: 对代码进行审查，发现潜在问题、安全漏洞和性能瓶颈
category: developer-tools
tags: [code, review, qa]
prompt_template:
template: |
你是一个资深代码审查专家...
input_variables:
- name: language
type: string
required: true
- name: code_content
type: string
required: true
output_schema:
type: object
properties:
critical_count: integer
warning_count: integer
suggestion_count: integer
issues: array
summary: string
execution:
engine: chat_completion
model:
provider: openai
name: gpt-4o
temperature: 0.3
retry:
max_attempts: 3
backoff: exponential
timeout: 30000
4.3 Skill设计原则
设计一个好的Skill，需要遵循以下原则。
原则1：单一职责
一个Skill只做一件事，并且做好这一件事。
反面案例：
# 不好的Skill：职责太多
name: code_assistant
description: 能写代码、审查代码、生成文档、写测试
正面案例：
# 好的Skill：每个职责一个Skill
name: code_review # 只管审查
name: code_generation # 只管代码生成
name: doc_generation # 只管文档
name: test_generation # 只管测试
为什么？
- 单一职责的Skill更容易测试
- 可以灵活组合（"先审查，再生成文档"）
- 可以单独优化（只升级code_review，不影响其他）
- 更容易被复用场景发现
原则2：可组合
Skill之间应该能像乐高一样拼装。
好的设计：
# 验证输入 → 处理 → 格式化输出
input_validator → core_processor → output_formatter
错误的设计：
# 把三个步骤塞进一个Skill
monolithic_skill:
does_everything: true # 无法复用其中任何一步
组合方式有三种：
- 顺序组合（Chain）：A的输出 → B的输入
- 条件组合（Router）：根据条件选择不同的Skill
- 并行组合（Fan-out）：多个Skill同时执行
原则3：可配置
Skill应该通过参数调整行为，而不是修改模板。
# 好的设计：通过参数控制
name: summarize
parameters:
length: short # short/medium/detail
style: bullet # bullet/paragraph/structure
language: zh # 输出语言
# 坏的实践：每个变体一个Skill
name: summarize_short_bullet # ❌ 不应该
name: summarize_medium_para # ❌ 不应该
配置参数的类型：
| 参数类型 | 说明 | 示例 |
|---|---|---|
| 内容参数 | 影响输出内容 | length, style |
| 格式参数 | 影响输出格式 | output_format, include_examples |
| 行为参数 | 影响执行方式 | temperature, max_tokens |
| 上下文参数 | 影响知识范围 | context_window, reference_docs |
原则4：有输入输出Schema
每个Skill必须有明确的输入输出定义。
input_schema:
type: object
properties:
code:
type: string
description: "待审查的代码"
required: true
max_length: 8000
language:
type: string
enum: [python, javascript, go, java, rust]
default: python
output_schema:
type: object
properties:
score:
type: integer
minimum: 0
maximum: 100
issues:
type: array
items:
type: object
properties:
line: integer
severity: string
message: string
为什么Schema重要？
- 输入验证：提前发现参数错误，不浪费AI调用
- 输出解析：结构化结果方便后续程序处理
- 自动补全：IDE可以和Agent框架根据Schema生成调用代码
- 类型安全：编译期就能检查参数类型错误
4.4 编写你的第一个Skill
实战：创建一个天气查询Skill。
Step 1: 定义元数据
# skills/weather-query.yaml
name: weather_query
version: 1.0.0
author: iceinto
description: 查询指定城市的天气信息，包括温度、湿度、风力等
category: information-retrieval
tags: [weather, tool-use, realtime]
Step 2: 设计输入输出Schema
input_schema:
type: object
properties:
city:
type: string
description: "城市名称，支持中文和英文，如'北京'、'Shanghai'"
required: true
days:
type: integer
description: "预报天数，1-7天"
default: 1
minimum: 1
maximum: 7
unit:
type: string
enum: [celsius, fahrenheit]
default: celsius
output_schema:
type: object
properties:
city: string
current:
temperature: number
humidity: number
wind_speed: number
condition: string
feels_like: number
forecast:
type: array
items:
date: string
high: number
low: number
condition: string
Step 3: 编写Prompt Template
这个Skill需要让AI具备"工具调用"能力——AI不是直接知道天气，而是需要生成工具调用指令，由Engine去获取真实数据。
prompt_template:
template: |
你现在是一个天气助手。用户需要查询天气信息。
如果你需要获取实时数据，请使用以下工具：
## 可用工具
{tools_description}
## 用户需求
城市：{city}
预报天数：{days}
温度单位：{unit}
## 执行规则
1. 先调用天气API获取数据
2. 根据获取到的数据组织回答
3. 回答要简洁、信息完整
4. 如果城市名称不确定，向用户确认
input_variables:
- name: city
type: string
required: true
- name: days
type: integer
default: 1
- name: unit
type: string
default: celsius
- name: tools_description
type: string
description: "由引擎自动注入的工具描述"
output_schema:
type: object
properties:
reply: string
tool_calls:
type: array
items:
tool: string
args: object
Step 4: 配置执行参数
execution:
engine: tool_use # 工具调用模式
model:
provider: openai
name: gpt-4o
temperature: 0.1 # 天气查询需要低随机性
tools:
- name: get_weather
description: "获取城市实时天气"
parameters:
city: string
days: integer
retry:
max_attempts: 2
backoff: fixed
interval: 1000
timeout: 15000
Step 5: 使用这个Skill
在Agent中调用：
# agent调用weather_query skill
result = agent.execute_skill("weather_query", {
"city": "上海",
"days": 3,
"unit": "celsius"
})
print(result.reply)
# "上海今天（5月8日）多云，24-28°C，东南风3-4级，湿度65%。未来两天有小雨..."
完整的Skill，就是这样一个从"定义"到"使用"的闭环。
4.5 Skill-Engine架构
Skill-Engine是驱动Skill运行的"发动机"。它负责：加载Skill、解析模板、注入变量、调用AI、处理结果。
核心架构图
┌─────────────────────────────────────────────────────────────┐
│ Skill-Engine │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Loader │ │ Parser │ │ Executor │ │ Post- │ │
│ │ (加载器) │───→│ (解析器) │───→│ (执行器) │───→│ processor │ │
│ │ │ │ │ │ │ │ (后处理) │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Skill库 │ │ 上下文 │ │ LLM调用 │ │ 结果解析 │ │
│ │ (存储) │ │ Manager │ │ (网络/API)│ │ (结构化) │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘
执行流程
一个Skill从被触发到返回结果的完整流程：
输入参数 {"city": "北京", "days": 1}
│
▼
┌─────────────────┐
│ 1. Skill解析器 │ 读取Skill定义（YAML/JSON）
│ 解析元数据 │ 验证Schema，检查参数完整性
│ 校验输入 │
└─────────────────┘
│
▼
┌─────────────────┐
│ 2. 模板引擎 │ 将city=北京, days=1
│ 变量替换 │ 注入到Prompt模板中
│ 准备Prompt │
└─────────────────┘
│
▼
┌─────────────────┐
│ 3. 上下文管理 │ 加载对话历史
│ 注入上下文 │ 加载相关工具描述
│ 组装完整请求 │
└─────────────────┘
│
▼
┌─────────────────┐
│ 4. 执行器 │ 调用LLM API
│ 调用LLM │ 处理重试逻辑
│ 处理超时 │
└─────────────────┘
│
▼
┌─────────────────┐
│ 5. 输出解析器 │ 解析LLM返回值
│ 格式转换 │ 转换为结构化数据
│ 验证输出Schema │ 校验输出Schema
└─────────────────┘
│
▼
返回结果 {temperature: 26, humidity: 60%, ...}
关键组件详解
1. Loader（加载器）
负责从Skill库加载Skill定义。
class SkillLoader:
def __init__(self, skill_repository: SkillRepository):
self.repository = skill_repository
def load(self, skill_name: str, version: str = None) -> SkillDefinition:
"""加载Skill定义"""
# 1. 从存储中获取Skill定义
raw = self.repository.fetch(skill_name, version)
# 2. 解析YAML/JSON为对象
definition = SkillDefinition.parse(raw)
# 3. 验证定义完整性
self._validate(definition)
return definition
def _validate(self, definition: SkillDefinition):
"""验证Skill定义必须字段"""
required = ["name", "version", "prompt_template.template"]
for field in required:
if not definition.has_field(field):
raise SkillValidationError(f"缺少必填字段: {field}")
2. Template Engine（模板引擎）
负责将变量注入到Prompt模板中。
class TemplateEngine:
def render(self, template: str, variables: dict) -> str:
"""渲染模板：替换变量 + 处理条件逻辑"""
rendered = template
# 1. 基础变量替换
for key, value in variables.items():
placeholder = "{" + key + "}"
rendered = rendered.replace(placeholder, str(value))
# 2. 检查是否有未替换的变量
unmatched = self._find_unmatched(rendered)
if unmatched:
raise TemplateRenderError(f"未替换的变量: {unmatched}")
return rendered
def _find_unmatched(self, text: str) -> list:
"""查找模板中残留的 {variable} 模式"""
import re
return re.findall(r"\{(\w+)\}", text)
3. Context Manager（上下文管理器）
管理每次Skill执行的环境信息。
class ContextManager:
def __init__(self):
self.session_store = {}
def build_context(self, skill: SkillDefinition, params: dict) -> ExecutionContext:
"""构建执行上下文"""
context = ExecutionContext()
# 会话上下文（对话历史）
context.session = self._get_session(params.get("session_id"))
# 全局上下文（系统配置、用户信息）
context.global_config = self._load_global_config()
# 工具上下文（可用工具描述）
if skill.execution.tools:
context.tools = self._describe_tools(skill.execution.tools)
# 安全上下文（权限、速率限制）
context.security = SecurityContext(
user_id=params.get("user_id"),
permissions=params.get("permissions", []),
)
return context
4. Executor（执行器）
核心执行引擎，负责任务调度和LLM调用。
class SkillExecutor:
def __init__(self, llm_client: LLMClient):
self.llm = llm_client
async def execute(self, context: ExecutionContext) -> SkillResult:
"""执行Skill"""
start_time = time.time()
for attempt in range(context.skill.execution.retry.max_attempts):
try:
# 调用LLM
response = await self.llm.chat_completion(
model=context.skill.execution.model.name,
messages=[
{"role": "system", "content": context.prompt},
*context.session.messages,
],
temperature=context.skill.execution.model.temperature,
max_tokens=context.skill.execution.model.max_tokens,
tools=context.tools if hasattr(context, 'tools') else None,
timeout=context.skill.execution.timeout,
)
# 后处理
result = await self._post_process(response, context)
# 记录执行指标
result.metadata = {
"latency_ms": int((time.time() - start_time) * 1000),
"attempts": attempt + 1,
"model": context.skill.execution.model.name,
}
return result
except RetryableError as e:
if attempt == context.skill.execution.retry.max_attempts - 1:
raise
await asyncio.sleep(self._backoff(attempt, context))
continue
def _backoff(self, attempt: int, context: ExecutionContext) -> float:
"""计算退避时间"""
strategy = context.skill.execution.retry.backoff
if strategy == "exponential":
return min(2 ** attempt * 1, 30) # 1, 2, 4, 8...秒
elif strategy == "fixed":
return context.skill.execution.retry.interval / 1000
return 0
4.6 Skill库的组织
随着Skill数量的增长，你需要一套良好的组织规范。
目录结构
skills/
├── official/ # 官方Skill（内置）
│ ├── code/
│ │ ├── code-review.yaml
│ │ ├── code-generation.yaml
│ │ └── test-generation.yaml
│ ├── text/
│ │ ├── translate.yaml
│ │ ├── summarize.yaml
│ │ └── grammar-check.yaml
│ └── data/
│ ├── json-to-csv.yaml
│ ├── data-extract.yaml
│ └── format-convert.yaml
│
├── community/ # 社区贡献Skill
│ ├── weather-query.yaml
│ ├── news-search.yaml
│ └── stock-analysis.yaml
│
├── custom/ # 自定义Skill（当前项目）
│ ├── pr-review.yaml
│ ├── release-note.yaml
│ └── api-doc-gen.yaml
│
├── tests/ # Skill测试
│ ├── test_weather_query.yaml
│ ├── expected_outputs/
│ │ ├── weather_query_sample.json
│ │ └── code_review_sample.json
│ └── fixtures/
│ └── sample_code.py
│
└── skill-registry.yaml # 注册表（索引全部Skill）
命名规范
| 规范 | 说明 | 示例 |
|---|---|---|
| 文件名 | 小写+中划线 | code-review.yaml |
| Skill名称 | 下划线分隔 | code_review |
| 分类 | 两级：大类/子类 | code/generation |
| 版本 | 语义化版本 | 1.2.0 |
skill-registry.yaml
# skill-registry.yaml
version: 1
updated: "2026-05-08"
skills:
- name: code_review
path: official/code/code-review.yaml
version: 1.2.0
category: code
tags: [code, review]
checksum: "a3f8b2c1..."
- name: translate
path: official/text/translate.yaml
version: 2.0.0
category: text
tags: [translation, nlp]
- name: weather_query
path: community/weather-query.yaml
version: 1.0.0
category: data
tags: [weather, tool-use]
依赖管理
Skill之间可能存在依赖关系。
# 一个复杂的Skill可能依赖其他Skill
name: pr_review_report
version: 1.0.0
dependencies:
- name: code_review
version: ">=1.0.0"
- name: summarize
version: ">=2.0.0"
- name: format_convert
version: "~1.2.0"
# Engine加载时，会先检查依赖是否满足
# 然后按依赖顺序加载所有Skill
依赖冲突处理：
| 策略 | 说明 | 适用场景 |
|---|---|---|
| 版本锁定 | 固定依赖版本 | 生产环境，确定性优先 |
| 版本范围 | 允许一定范围的版本浮动 | 开发环境，灵活性优先 |
| 沙箱隔离 | 不同Skill用不同版本 | 复杂项目，兼容性优先 |
4.7 Skill的测试与调试
Skill的质量直接影响Agent的可靠性。你需要一套测试体系。
测试层级
┌──────────────────────────────────────┐
│ 3. 集成测试 │
│ Multi-Skill Chain 测试 │
│ 真实LLM调用（可选） │
├──────────────────────────────────────┤
│ 2. 单元测试 │
│ 输入输出Schema验证 │
│ 模板渲染测试 │
│ 边界条件测试 │
│ Mock LLM返回值 │
├──────────────────────────────────────┤
│ 1. 静态检查 │
│ YAML格式校验 │
│ 必填字段检查 │
│ 变量一致性检查 │
└──────────────────────────────────────┘
1. 静态检查
def validate_skill_definition(yaml_content: str) -> list:
"""检查Skill定义文件的正确性"""
errors = []
# 解析YAML
try:
skill = yaml.safe_load(yaml_content)
except yaml.YAMLError as e:
return [f"YAML解析错误: {e}"]
# 必填字段检查
required = ["name", "version", "prompt_template"]
for field in required:
if field not in skill:
errors.append(f"缺少必填字段: {field}")
# 模板变量和input_variables一致性检查
if "prompt_template" in skill:
template = skill["prompt_template"].get("template", "")
used_vars = set(re.findall(r"\{(\w+)\}", template))
declared_vars = {v["name"] for v in
skill["prompt_template"].get("input_variables", [])}
for var in used_vars - declared_vars:
errors.append(f"模板中使用了变量 '{var}' 但未在input_variables中声明")
for var in declared_vars - used_vars:
errors.append(f"input_variables声明了 '{var}' 但模板中未使用")
# 版本格式检查
if "version" in skill:
if not re.match(r"^\d+\.\d+\.\d+$", skill["version"]):
errors.append("版本格式错误，应为 semver 格式 (x.y.z)")
return errors
2. 单元测试
# tests/test_weather_query.py
import pytest
from skill_engine import TemplateEngine, SkillLoader
from unittest.mock import AsyncMock
class TestWeatherQuerySkill:
"""天气查询Skill单元测试"""
def setup_method(self):
self.loader = SkillLoader()
self.skill = self.loader.load("weather_query")
self.engine = TemplateEngine()
def test_template_rendering(self):
"""测试模板渲染：正确的变量替换"""
result = self.engine.render(
self.skill.prompt_template.template,
{"city": "北京", "days": 1, "tools_description": "...", "unit": "celsius"}
)
assert "北京" in result
assert "1" in result or "一天" in result
def test_input_validation_missing_required(self):
"""测试输入校验：缺少必填参数"""
with pytest.raises(ValidationError):
self.skill.validate_input({"days": 3}) # 缺少 city
def test_input_validation_invalid_days(self):
"""测试输入校验：超出范围的参数"""
with pytest.raises(ValidationError):
self.skill.validate_input({"city": "北京", "days": 10}) # 最多7天
@pytest.mark.asyncio
async def test_execution_with_mock_llm(self):
"""测试执行流程：Mock LLM调用"""
# Mock LLM返回假数据
mock_llm = AsyncMock()
mock_llm.chat_completion.return_value = {
"choices": [{"message": {"content": '{"temperature": 26}'}}]
}
executor = SkillExecutor(mock_llm)
result = await executor.execute(
skill=self.skill,
params={"city": "北京", "days": 1}
)
assert result.data["temperature"] == 26
mock_llm.chat_completion.assert_called_once()
3. 测试数据示例
# tests/expected_outputs/weather_query_sample.json
{
"city": "北京",
"current": {
"temperature": 26,
"humidity": 55,
"wind_speed": 12,
"condition": "晴",
"feels_like": 27
},
"forecast": [
{
"date": "2026-05-08",
"high": 28,
"low": 18,
"condition": "晴转多云"
}
]
}
调试技巧
| 问题 | 检查点 | 修复方法 |
|---|---|---|
| 输出格式不对 | 检查output_schema和模板中的格式描述 | 在模板中明确指定输出格式 |
| 变量没有替换 | 检查模板中{var_name}拼写 | 使用_find_unmatched检查 |
| AI输出不稳定 | 检查temperature设置 | 降低temperature，增加Few-shot示例 |
| 执行超时 | 检查模型和token限制 | 减少max_tokens，简化模板 |
| 依赖冲突 | 检查skill-registry版本 | 锁定版本或使用沙箱隔离 |
本章小结
| 要点 | 内容 |
|---|---|
| 什么是Skill | 可复用的AI能力单元，包含元数据、模板、执行逻辑 |
| Skill结构 | Metadata（身份证）+ Prompt Template（大脑）+ Execution Logic（执行器） |
| 设计原则 | 单一职责、可组合、可配置、有输入输出Schema |
| Skill-Engine | 加载器→解析器→模板引擎→执行器→后处理器 |
| Skill库组织 | 目录分类 + 命名规范 + 注册表 + 依赖管理 |
| 测试体系 | 静态检查 → 单元测试 → 集成测试，逐层保障质量 |
思考与练习
-
分析：找出你日常工作中一个可以封装为Skill的重复性任务。它的输入是什么？输出是什么？模板怎么写？
-
动手：按照4.4节的步骤，编写一个"会议纪要生成Skill"。输入是会议录音转文字文本，输出是结构化的会议纪要（议题、结论、待办事项、负责人）。
-
设计：给"会议纪要生成Skill"添加参数——
style（正式/简洁）、language（中文/英文）、include_action_items（是否包含待办事项）。怎么设计Schema和模板？ -
架构：如果两个Skill需要共享同一个上下文（比如"代码审查"和"生成测试"都针对同一段代码），你会怎么设计上下文传递机制？
-
思考：Skill能否嵌套调用？比如"生成PR描述"Skill调用"代码审查"Skill的结果作为输入。这种链式调用的优缺点是什么？
下一章：第5章：Multi-Agent协作模式
第5章：Multi-Agent协作模式
学习本章后，你将：理解Multi-Agent的适用场景，掌握常见协作模式，能够设计和实现多Agent系统。
5.1 为什么需要Multi-Agent
一个Agent不够用
当你让ChatGPT写一篇长文章，你会发现：写开头忘了结尾要求，写技术部分不够专业，写完后还要自己检查错误。单个Agent有固有的局限性——它擅长单一领域生成任务，但不擅长多角色协作、长时间连贯执行、专业分工。
三个核心驱动力
1. 复杂任务分解
开餐厅需要招商经理、设计师、厨师、会计——你不会一个人做所有事。Multi-Agent同理：把复杂任务拆成子任务，每个Agent负责自己擅长的部分。
2. 专业分工
| Agent角色 | 擅长的事 | 不擅长的事 |
|---|---|---|
| Coder Agent | 写代码、调试 | 设计架构、写文档 |
| Reviewer Agent | 找Bug、代码规范 | 理解业务需求 |
| Tester Agent | 写测试、边界覆盖 | 做UI设计 |
| Architect Agent | 系统设计、选型 | 写具体实现 |
一个全能Agent = 每个领域平均水平的Agent。多个专业Agent = 每个领域都有专家。
3. 并行执行
单Agent： Task A → B → C → D = 40分钟（串行）
Multi-Agent： Task A → Agent 1 ──┐
Task B → Agent 2 ──┤ 并行 = 15分钟
Task C → Agent 3 ──┤
Task D → Agent 4 ──┘
5.2 常见协作模式
5.2.1 主从模式（Master-Worker）
结构：一个协调者分发任务，多个Worker执行，协调者汇总结果。
┌──────────────────────┐
│ Coordinator │
│ 接收、分解、汇总 │
└──────────┬───────────┘
│
┌─────────────────────┼─────────────────────┐
│ │ │
┌─────▼──────┐ ┌─────▼──────┐ ┌─────▼──────┐
│ Worker 1 │ │ Worker 2 │ │ Worker 3 │
│ 数据分析 │ │ 代码生成 │ │ 文档生成 │
└────────────┘ └────────────┘ └────────────┘
流程：接收任务 → 分解子任务 → 分发Worker → 并行执行 → 汇总结果 → 最终输出
典型案例：AI编程助手（协调者调度代码生成、审查、测试三个Agent）
5.2.2 对等模式（Peer-to-Peer）
结构：所有Agent地位平等，可直接通信讨论，没有中心节点。
┌──────────────────┐
│ Agent A │
│ 架构师角色 │
└──────┬───────────┘
│
┌────────────────┼────────────────┐
│ │ │
┌─────▼──────┐ ┌─────▼──────┐ ┌─────▼──────┐
│ Agent B │◄─┤ Agent C │◄─┤ Agent D │
│ 前端角色 │──►│ 后端角色 │──►│ QA角色 │
└────────────┘ └────────────┘ └────────────┘
流程：Agent提出方案 → 各Agent评估讨论 → 达成一致 → 各自执行 → 同步结果
典型案例：AI辩论系统（多个Agent扮演不同角色讨论议题）
5.2.3 流水线模式（Pipeline）
结构：任务按固定顺序经过多个Agent，每个处理一个环节后传递给下一个。
输入 → ┌──────────┐ → ┌──────────┐ → ┌──────────┐ → ┌──────────┐ → 输出
│ Agent A │ │ Agent B │ │ Agent C │ │ Agent D │
│ 需求分析 │ │ 架构设计 │ │ 代码实现 │ │ 测试验证 │
└──────────┘ └──────────┘ └──────────┘ └──────────┘
特点：每个Agent的输入是上一个的输出，链式处理，职责明确。
典型案例：文档生成流水线（需求分析 → 内容生成 → 格式美化 → 校对审核）
5.2.4 树状模式（Hierarchical）
结构：多层级管理，每层Agent管理下一层，形成树状层级。
┌──────────────────┐
│ Super Agent │
└────────┬─────────┘
│
┌─────────────┼─────────────┐
│ │ │
┌──────▼──────┐ ┌───▼───────┐ ┌───▼───────┐
│ Team Lead A │ │Team Lead B│ │Team Lead C│
│ 技术团队 │ │ 业务团队 │ │ 运营团队 │
└──────┬──────┘ └───┬───────┘ └───┬───────┘
│ │ │
┌────┼────┐ ┌───┼───┐ ┌────┼────┐
│ │ │ │ │ │ │ │ │
Dev1 Dev2 .. Biz1 Biz2 .. Ops1 Ops2 ..
典型案例：大规模软件开发（Super PM → 多个Team Lead → 多个Developer）
模式对比
| 模式 | 控制方式 | 优点 | 缺点 | 最佳场景 |
|---|---|---|---|---|
| 主从模式 | 中心化 | 控制简单、职责清晰、易监控 | 单点故障、中心瓶颈 | 任务分发、代码生成 |
| 对等模式 | 去中心化 | 高可用、讨论充分 | 协调复杂、消息风暴 | 方案评审、创意讨论 |
| 流水线模式 | 顺序控制 | 结构简单、职责明确、易调试 | 最慢环节限制整体 | 数据处理、CI/CD |
| 树状模式 | 层级控制 | 扩展性强、管理粒度细 | 延迟高、通信开销大 | 大型项目、多团队协作 |
5.3 Agent通信协议
消息格式（Message Schema）
Agent之间需要统一的"语言"来通信。
{
"message_id": "msg_20240301_001",
"sender": { "agent_id": "coder_1", "role": "Coder" },
"receiver": { "agent_id": "reviewer_1", "role": "Reviewer" },
"message_type": "task_result",
"content": { "task_id": "task_001", "status": "completed" },
"context": { "parent_message_id": "msg_20240301_000", "session_id": "s1" },
"timestamp": "2024-03-01T10:30:00Z",
"ttl": 300000
}
核心字段：
| 字段 | 用途 |
|---|---|
| message_id | 唯一标识，用于追踪、去重、重试 |
| sender/receiver | 消息路由，谁发的、发给谁 |
| message_type | 类型：task_assign / task_result / review / question / error |
| content | 实际传递的数据 |
| context | 关联的上游消息和会话ID |
| ttl | 生存时间，超时未处理则丢弃 |
消息路由
三种路由方式：
1. 直接路由 ── 指定receiver，点对点发送（coder_1 → reviewer_1）
2. 广播路由 ── 所有Agent接收（coordinator → all_agents）
3. 主题路由 ── 按主题订阅（agent订阅 "task.coding.*"）
异步处理
Agent通信必须异步，不要阻塞等待：
同步（不要用）：Agent A → B，A等待响应，不能做其他事
异步（推荐）： Agent A → Message Queue → Agent B，A立即返回继续工作
← 回调通知 ←
消息队列作用：解耦、缓冲、重试、追踪
5.4 状态共享与隔离
分层状态
┌──────────────────────────────────────────────────┐
│ 全局上下文（只读） │
│ 所有Agent可读：项目需求、架构决策、全局配置 │
├──────────────────────────────────────────────────┤
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Agent A │ │ Agent B │ │ Agent C │ │
│ │ 私有状态 │ │ 私有状态 │ │ 私有状态 │ │
│ │ 当前任务 │ │ 当前代码 │ │ 当前测试 │ │
│ └────────────┘ └────────────┘ └────────────┘ │
└──────────────────────────────────────────────────┘
| 层级 | 内容 | 访问权限 |
|---|---|---|
| 全局上下文 | 项目需求、架构决策、全局变量 | 所有Agent只读，协调者写入 |
| 会话上下文 | 当前任务信息、中间结果 | 相关Agent读写 |
| 私有上下文 | Agent临时变量、本地缓存 | 仅本Agent读写 |
状态同步策略
| 策略 | 描述 | 适用场景 |
|---|---|---|
| 全量同步 | 每次发送完整状态 | Agent少、状态小 |
| 增量同步 | 只同步变化部分 | Agent多、状态大 |
| 按需同步 | Agent需要时才请求 | 对延迟不敏感 |
| 事件驱动 | 状态变化时广播事件 | 实时协作系统 |
经验法则：尽可能让Agent无状态。状态越少，系统越稳定。
5.5 错误处理与重试
三级容错机制
单个Agent的失败不应该导致整个系统崩溃。
第一级：Agent内部重试
Agent执行任务 → 成功则返回结果
→ 失败 → 重试（1s→3s→放弃）
→ 仍失败 → 发送错误给协调者
第二级：协调者容错
收到Agent失败通知 → 重试：重新分配给另一个同类型Agent
→ 降级：使用简化版处理
→ 跳过：非关键任务直接跳过
第三级：系统级容错
| 机制 | 说明 |
|---|---|
| 超时控制 | 给每个Agent设置执行超时，不无限等待 |
| 熔断机制 | 连续失败N次的Agent被暂时隔离 |
| 健康检查 | 定期检查Agent状态，不可用的不参与分配 |
| 结果校验 | 关键结果做二次验证（审查 + 测试） |
幂等性设计
Agent必须做到幂等——同一个任务执行多次，结果一致。
非幂等（危险）：执行"插入一条数据" → 重试3次 → 插入3条数据
幂等（安全）： 执行"id已存在则跳过" → 重试3次 → 只有1条数据
原则：每个任务有唯一ID，Agent记录已执行任务ID，重复ID直接返回缓存结果。
5.6 实战：搭建一个Multi-Agent编程系统
系统设计
┌─────────────────────────────┐
│ Coordinator │
│ 负责任务分解和结果汇总 │
└────────────┬────────────────┘
│
┌───────────────────────┼───────────────────────┐
│ │ │
┌────▼────┐ ┌─────▼─────┐ ┌─────▼────┐
│ Coder │ │ Reviewer │ │ Tester │
│ 生成代码 │ │ 审查代码 │ │ 测试代码 │
└─────────┘ └───────────┘ └──────────┘
流程：接收任务 → Coder生成代码 → Reviewer审查 → 不通过则修改 → Tester测试 → 汇总输出
核心实现（简化代码）
class Message:
def __init__(self, sender, receiver, type, content):
self.sender, self.receiver = sender, receiver
self.type, self.content = type, content
class Coordinator:
def __init__(self):
self.agents = {}
def send(self, msg):
self.agents[msg.receiver].queue.append(msg)
def pop(self, aid):
return self.agents[aid].queue.pop().content
def run(self, task):
# 分发编码任务
self.send(Message("coord", "coder", "code", task))
code = self.pop("coder")["code"]
# 审查代码
self.send(Message("coord", "reviewer", "review", {"code": code}))
review = self.pop("reviewer")
if not review["approved"]:
self.send(Message("coord", "coder", "fix",
{"code": code, "feedback": review["comments"]}))
code = self.pop("coder")["code"]
# 测试验证
self.send(Message("coord", "tester", "test", {"code": code}))
test = self.pop("tester")
return {"code": code, "review": review, "test": test}
class Agent:
def __init__(self, id): self.id = id; self.queue = []
def handle(self, msg): pass
class CoderAgent(Agent):
def handle(self, msg):
code = "# 实现：" + msg.content.get("description", "")
self.queue.append(Message(self.id, "coord", "result", {"code": code}))
class ReviewerAgent(Agent):
def handle(self, msg):
self.queue.append(Message(self.id, "coord", "result",
{"approved": True, "comments": []}))
class TesterAgent(Agent):
def handle(self, msg):
self.queue.append(Message(self.id, "coord", "result",
{"passed": True, "coverage": 85.0}))
# 启动
coord = Coordinator()
agents = [CoderAgent("coder"), ReviewerAgent("reviewer"), TesterAgent("tester")]
for a in agents: coord.agents[a.id] = a
result = coord.run("实现用户管理系统")
print(result)
关键设计要点
| 设计点 | 实现 | 为什么重要 |
|---|---|---|
| 异步通信 | 消息队列 + 路由 | 避免Agent互相阻塞 |
| 消息驱动 | 统一Message结构 | 所有Agent用同一种语言通信 |
| 重试机制 | max_retries + 超时 | 单个Agent失败不影响整体 |
| 结果汇总 | Coordinator统一收集 | 控制最终输出质量 |
| 模块化 | 每个Agent独立可替换 | 方便升级和A/B测试 |
扩展方向：加入健康检查自动替换失败Agent；加入结果缓存减少重复调用；加入监控面板可视化Agent执行流程；加入人工审核节点处理关键决策。
本章小结
| 要点 | 内容 |
|---|---|
| 为什么需要Multi-Agent | 复杂任务分解、专业分工、并行执行，突破单个Agent能力边界 |
| 4种协作模式 | 主从（中心分发）、对等（平等通信）、流水线（链式处理）、树状（层级管理） |
| 通信协议 | 统一消息格式，支持直接/广播/主题路由，异步消息队列 |
| 状态管理 | 分层：全局只读 + 会话共享 + 私有隔离，优先保证无状态 |
| 错误处理 | 三级容错：Agent内重试 → 协调者调度 → 系统级熔断/降级 |
| 实战架构 | Coordinator调度Coder + Reviewer + Tester，消息驱动协作 |
思考与练习
- 选择：你的业务场景适合哪种协作模式？用对比表的维度评估一下
- 设计：画一个你当前系统的Multi-Agent架构图，标注通信方式和状态管理方案
- 实现：基于5.6节代码，用你熟悉的语言实现两个Agent协作："一个生成Markdown文档，另一个审查格式规范"
- 思考：如果Reviewer和Coder产生分歧，你的系统应该怎么处理？
下一章：第6章：企业落地：需求→AI生成→自动PR
第6章：企业落地：需求→AI生成→自动PR
学习本章后，你将：理解企业如何落地AI-Agent系统，掌握从需求到代码的全链路自动化，能够规划和实施Agent转型。
6.1 企业落地三步走
很多CTO对AI Agent的期待是"买一个工具，所有人自动提效"。这是最大的误区。AI Agent是能力建设，不是SaaS产品。
正确路径：试点 → 推广 → 规模化
落地阶段 关键动作
┌────────────────────────────────────────────────────────────────┐
│ 第1步：试点（1-2月） │
│ ├── 选一个合适的团队/项目 │
│ ├── 跑通"需求→方案→代码→PR"全链路 │
│ ├── 积累模板和最佳实践 │
│ └── 产出可量化的效果数据 │
├────────────────────────────────────────────────────────────────┤
│ 第2步：推广（3-6月） │
│ ├── 推广到3-5个团队 │
│ ├── 建立内部Prompt模板库 │
│ ├── 培养"Agent引导师"（每团队1-2人） │
│ └── 迭代优化Agent执行流程 │
├────────────────────────────────────────────────────────────────┤
│ 第3步：规模化（6-12月） │
│ ├── 全技术部门接入 │
│ ├── 形成组织级Agent平台 │
│ ├── 建立质量门禁和度量体系 │
│ └── 持续优化，形成飞轮效应 │
└────────────────────────────────────────────────────────────────┘
试点团队选取标准
| 条件 | 为什么重要 |
|---|---|
| 成员对AI有好奇心 | 愿意试错，不抵触新工具 |
| 有标准化开发流程 | Agent的产出容易被评估 |
| 项目有明确的交付周期 | 效果可以量化对比 |
| 技术负责人支持 | 上层推动，减少阻力 |
我在金仕达的第一个试点是量化交易系统内部工具开发团队。原因：需求标准化（"写数据同步脚本"、"生成行情接口"），且团队对新技术的接受度最高。试点第一周，交付速度就提升了3倍。
6.2 需求理解：从业务语言到Agent任务
企业落地最大的鸿沟不是技术，而是需求转化。
业务人员说： "给我做一个风控报表"
↓
技术经理理解： "需要一个可配置的风控数据看板"
↓
Agent需要： "用React+ECharts实现风控Dashboard，
数据源来自风控系统REST API，
包含5个核心KPI的实时展示"
Agent比人类更"较真"——它需要精确指令才能精确输出。
需求转化三步法
Step 1：提取关键要素
| 业务要素 | Agent指令 |
|---|---|
| "做个报表" | 输出格式（Dashboard/PDF/Excel） |
| "展示风控数据" | 数据来源、字段定义 |
| "要实时的" | 刷新频率（秒级/分钟级） |
Step 2：补充隐含信息
业务说不出来但Agent不能不知道的：
业务说："给银行做一个量化交易Agent"
隐含信息：
- 交易品种：贵金属（黄金、白银）
- 市场：上海黄金交易所
- 合规要求：交易前风险检查
- 风控规则：单笔最大交易量、日累计限额
- 系统架构：对接金仕达交易系统
Step 3：拆解为Agent任务树
需求："量化交易Agent"
├── 行情数据接入（上金所行情源、数据清洗存储）
├── 策略执行引擎（策略配置接口、交易信号生成）
├── 风控模块（交易前检查、实时监控告警）
└── 交易记录与对账（交易流水、每日报表）
Case：银行量化交易Agent需求分析
背景：某商业银行需要贵金属量化交易系统，接入上海黄金交易所。
业务人员描述："做一套自动交易的系统，根据行情变化自动买卖黄金，要能控制风险。"
需求转化后的Agent任务定义：
Agent任务定义模板
=================
项目名称：银行贵金属量化交易Agent
目标用户：银行交易员
核心场景：基于技术指标的自动化交易
1. 上金所行情接入模块
- 品种：Au99.99, Au(T+D), Ag(T+D)
- 频率：Tick级/1分钟/5分钟
2. 策略引擎
- 策略类型：均线策略、RSI策略、自定义
- 支持：JSON配置 / 历史回测
3. 风控模块
- 单笔限仓：100手，日累计：500手
- 止损线：-2%自动平仓
4. 订单管理
- 订单类型：市价单、限价单、止损单
- 对接银行核心交易系统
5. 报表模块
- 日报：交易汇总、盈亏统计
- 实时看板：持仓、浮盈浮亏
6.3 AI生成实现方案（Planner输出技术方案）
企业开发不是写个函数就完了——需要技术方案，包括架构设计、数据库、API、部署。有了AI Agent，我们可以让Planner Agent自动生成。
Planner核心流程
业务需求（结构化）
↓
Planner Agent 读取任务定义
↓
┌──────────────────┐
│ 方案生成引擎 │
│ 架构 / 技术选型 │
│ 模块 / 数据流 │
│ 接口 / 工作量 │
└──────────────────┘
↓
技术方案（结构化JSON）
↓
人工审核 + 修订
↓
Coder Agent 读取方案 → 生成代码
Prompt模板：需求→方案
你是[公司]的资深架构师，熟悉公司技术栈和架构规范。
需求定义：
{结构化需求}
约束条件：
- 技术栈：Java + Spring Cloud / Go + Gin
- 数据库：MySQL + Redis
- 部署：Kubernetes
- 安全：API需鉴权，敏感数据加密
输出要求：
1. 系统架构图（ASCII）
2. 模块划分和各模块职责
3. 数据库表设计（核心表）
4. API接口定义（RESTful）
5. 核心业务流程
6. 关键技术决策及理由
7. 预估开发工作量（人/天）
Output Schema（Planner输出结构）
Planner Agent的输出必须是结构化的，后续的Coder Agent才能准确读取：
{
"project_name": "银行贵金属量化交易Agent",
"version": "1.0.0",
"architecture": {
"pattern": "微服务架构",
"diagram_ascii": "
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 行情服务 │ │ 策略服务 │ │ 风控服务 │
│ Go+Gin │ │Java+Spring│ │ Go+Gin │
└────┬─────┘ └────┬─────┘ └────┬─────┘
└──────────────┼──────────────┘
┌─────┴─────┐
│ Kafka MQ │
└─────┬─────┘
┌─────┴─────┐
│ 订单服务 │
│ SpringCloud│
└───────────┘
",
"services": [
{"name": "行情服务", "lang": "Go", "resp": "接入上金所行情数据"},
{"name": "策略服务", "lang": "Java", "resp": "策略加载与信号计算"},
{"name": "风控服务", "lang": "Go", "resp": "交易前检查与限额管理"},
{"name": "订单服务", "lang": "Java", "resp": "订单路由与对账"}
]
},
"database": {
"engine": "MySQL 8.0",
"tables": [{
"name": "trade_order",
"fields": [
{"name": "id", "type": "bigint", "pk": true},
{"name": "instrument_id", "type": "varchar(20)"},
{"name": "direction", "type": "varchar(4)", "comment": "BUY/SELL"},
{"name": "quantity", "type": "decimal(18,4)"},
{"name": "price", "type": "decimal(18,4)"},
{"name": "status", "type": "varchar(20)"},
{"name": "created_at", "type": "datetime"}
]
}]
},
"api": {
"base_path": "/api/v1",
"endpoints": [
{"method": "POST", "path": "/orders", "desc": "提交交易订单"},
{"method": "GET", "path": "/positions", "desc": "查询持仓"}
]
},
"workload_estimate": {
"total": 45,
"breakdown": [
{"module": "行情接入", "days": 8},
{"module": "策略引擎", "days": 12},
{"module": "风控模块", "days": 10},
{"module": "订单管理", "days": 10},
{"module": "报表模块", "days": 5}
]
}
}
这个JSON方案就是"需求到代码"的桥梁。方案的质量决定了代码的质量——Planner输出模糊，Coder一定生成垃圾。
6.4 自动代码生成（基于方案生成代码）
方案→代码的Prompt模板
你是[公司][语言]开发工程师，请根据技术方案生成代码。
技术方案：
{Planner输出的JSON}
你负责的模块：[模块名]
需要实现的接口：[接口列表]
生成要求：
1. 遵循公司代码规范
2. 包含完整的方法注释和关键逻辑注释
3. 添加必要的错误处理和日志
4. 代码可直接编译通过
5. 生成对应的单元测试
输出结构：
- 每个文件单独输出，标明文件路径
- 包含依赖配置（如有新增）
Coder Agent多轮机制
单次生成很难完美，需要多轮迭代：
Round 1: 生成骨架代码 → 人工审核结构
Round 2: 填充核心逻辑 → 人工审核逻辑
Round 3: 补充异常处理 → 人工审核完整性
Round 4: 生成测试 → 运行测试，修复失败项
Final: 审查通过 → 进入PR流程
Reviewer Agent自动审查
代码生成后，Reviewer Agent自动进行审查：
Reviewer Agent Prompt
=====================
你是一个资深的代码审查专家，请审查以下代码。
审查维度：
1. 方案一致性：代码是否完全实现技术方案的要求
2. 代码质量：命名规范、重复代码、圈复杂度
3. 安全隐患：SQL注入、XSS、敏感数据泄露
4. 性能问题：N+1查询、未使用缓存
5. 可测试性：依赖注入、便于mock
审查结论：[通过/需修改/不通过]
每个问题标注严重等级：CRITICAL/MAJOR/MINOR
{
"review_summary": {
"status": "需修改",
"critical_issues": 1,
"major_issues": 3
},
"issues": [
{
"severity": "CRITICAL",
"file": "OrderService.java",
"line": 45,
"description": "交易金额未做精度校验，可能导致金额计算错误",
"suggestion": "添加BigDecimal.setScale(4, RoundingMode.HALF_UP)"
},
{
"severity": "MAJOR",
"file": "MarketDataService.java",
"line": 78,
"description": "数据库查询未添加缓存，每秒数千次写入导致性能瓶颈",
"suggestion": "添加Redis缓存，缓存时间30秒"
}
]
}
审查结果自动回传Coder Agent修复，形成"A → 审查 → 修复 → 再审"闭环。
6.5 自动化PR流程
架构设计
┌────────────────────────────────────────────────────────────────┐
│ Agent → PR 全链路 │
├────────────────────────────────────────────────────────────────┤
│ │
│ Planner Agent → 输出JSON方案 │
│ ↓ │
│ Coder Agent → 生成代码 │
│ ↓ │
│ Reviewer Agent → 自动审查 │
│ ↓ 审查通过 │
│ AutoPR Engine │
│ ├── 创建新分支 (git branch) │
│ ├── 提交代码 (git commit) │
│ ├── 创建PR (GitHub API) │
│ ├── 关联需求编号和标签 │
│ └── 触发CI/CD流水线 │
│ ↓ │
│ 开发者 ← 收到PR通知 → 审核 → 合并 → 部署 │
└────────────────────────────────────────────────────────────────┘
Python实现：class AutoPR
import os, subprocess, requests
from typing import List, Optional
from datetime import datetime
class AutoPR:
"""自动PR引擎：接收Agent生成的代码，自动创建Pull Request"""
def __init__(self, repo_path: str, github_token: str,
repo_owner: str, repo_name: str,
base_branch: str = "main"):
self.repo_path = repo_path
self.github_token = github_token
self.repo_owner = repo_owner
self.repo_name = repo_name
self.base_branch = base_branch
def create_branch(self, branch_name: str) -> bool:
try:
subprocess.run(["git", "checkout", self.base_branch],
cwd=self.repo_path, check=True, capture_output=True)
subprocess.run(["git", "pull", "origin", self.base_branch],
cwd=self.repo_path, check=True, capture_output=True)
subprocess.run(["git", "checkout", "-b", branch_name],
cwd=self.repo_path, check=True, capture_output=True)
return True
except subprocess.CalledProcessError:
return False
def commit_and_push(self, files: List[dict], msg: str) -> bool:
try:
for f in files:
fp = os.path.join(self.repo_path, f["path"])
os.makedirs(os.path.dirname(fp), exist_ok=True)
with open(fp, "w", encoding="utf-8") as fh:
fh.write(f["content"])
subprocess.run(["git", "add", "."],
cwd=self.repo_path, check=True)
subprocess.run(["git", "commit", "-m", msg],
cwd=self.repo_path, check=True)
subprocess.run(["git", "push", "origin", "HEAD"],
cwd=self.repo_path, check=True)
return True
except subprocess.CalledProcessError:
return False
def create_pull_request(self, title: str, body: str,
head_branch: str,
reviewers: Optional[List[str]] = None,
labels: Optional[List[str]] = None) -> Optional[str]:
headers = {
"Authorization": f"token {self.github_token}",
"Accept": "application/vnd.github.v3+json"
}
pr_body = f"""
## Agent自动生成的PR
> 生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
### 需求来源
{body}
### 回滚方案
```bash
git revert {head_branch}
```
### 审查Checklist
- [ ] 代码实现是否完整
- [ ] 安全隐患排查
- [ ] 测试是否全覆盖
"""
data = {"title": title, "body": pr_body,
"head": head_branch, "base": self.base_branch}
resp = requests.post(
f"https://api.github.com/repos/{self.repo_owner}/{self.repo_name}/pulls",
headers=headers, json=data)
if resp.status_code == 201:
pr = resp.json()
if labels:
requests.post(pr["issue_url"] + "/labels",
headers=headers, json={"labels": labels})
if reviewers:
requests.post(pr["url"] + "/requested_reviewers",
headers=headers, json={"reviewers": reviewers})
return pr["html_url"]
return None
def run(self, files: List[dict], requirement: str,
module: str) -> Optional[str]:
ts = datetime.now().strftime("%Y%m%d-%H%M%S")
branch = f"ai-agent/{module}/{ts}"
print(f"[AutoPR] 分支: {branch}")
if not self.create_branch(branch):
return None
if not self.commit_and_push(files,
f"[AI-Agent] {module}: {requirement[:80]}"):
return None
return self.create_pull_request(
title=f"[AI-Agent] {module}: {requirement[:60]}",
body=f"需求：{requirement}\n模块：{module}\n审查结论：通过",
head_branch=branch,
reviewers=["tech-lead", "senior-dev"],
labels=["ai-generated", module])
# 使用示例
if __name__ == "__main__":
autopr = AutoPR(
repo_path="/path/to/repo",
github_token=os.environ["GITHUB_TOKEN"],
repo_owner="bank-name",
repo_name="quant-trading-system"
)
generated_files = [
{"path": "market-data/service/market_data_service.go",
"content": "// Agent generated code..."}
]
url = autopr.run(
files=generated_files,
requirement="实现上金所行情数据接入服务",
module="market-data")
print(f"PR created: {url}")
CI/CD集成
PR创建后自动触发CI流水线：代码规范检查 → 单元测试 → 集成测试 → 安全扫描 → 构建验证。测试结果自动评论到PR，开发者收到通知进行最终审核。
金仕达实践中，这套流程将"需求到代码上线"的平均时间从3天缩短到4小时。核心变化不是代码写得快了，而是等待时间消失了——Agent在后台完成所有前置工作，开发者只需做最终审核。
6.6 效果度量
KPI体系
| 维度 | 指标 | 基准值 | 目标值 |
|---|---|---|---|
| 效率 | 需求到交付周期 | 3天 | <1天 |
| 效率 | 开发吞吐量（Story Point/周） | 40点 | 80点 |
| 效率 | 代码生成占比 | 0% | 40% |
| 质量 | Bug率（/千行） | 15个 | <5个 |
| 质量 | 代码审查一次通过率 | 60% | 85% |
| 质量 | 测试覆盖率 | 40% | 75% |
| 成本 | AI调用成本（/Story Point） | 0 | <50元 |
| 成本 | ROI | - | >10倍 |
金仕达案例：研发效能提升5-6倍
我在金仕达主导的AI Agent落地项目，真实数据如下：
试点团队（内部工具开发团队，8人）
====================================
落地前（2023Q4）：
周完成需求：12个
交付周期：3.2天
Bug率：13.2/千行
测试覆盖率：35%
落地后（2024Q2，Agent辅助3个月后）：
周完成需求：68个（5.7倍）
交付周期：0.4天（8倍）
Bug率：4.8/千行（降低63%）
测试覆盖率：72%（提升106%）
核心指标变化：
┌──────────────────────────────────────────┐
│ 指标 落地前 落地后 提升 │
├──────────────────────────────────────────┤
│ 周完成需求 12 68 5.7x │
│ 交付周期(天) 3.2 0.4 8.0x │
│ Bug率(/千行) 13.2 4.8 -63% │
│ 测试覆盖率 35% 72% +37pp │
│ 开发满意度(5分) 3.2 4.6 +44% │
└──────────────────────────────────────────┘
为什么能提升5-6倍？ 不是因为"AI写代码比人快6倍"，而是：
效率提升分解
┌──────────────────────────────────────┐
│ 方案自动生成 30% │
│ 消除等待时间 20% │
│ 代码生成加速 25% │
│ 自动代码审查 15% │
│ 自动PR流程 10% │
├──────────────────────────────────────┤
│ 综合效率提升 5-6倍 │
└──────────────────────────────────────┘
成本分析
月度成本（8人团队）
API调用费用（GPT-4 + Claude 3）： ¥8,000
平台维护： ¥2,000
培训支持： ¥1,000
合计： ¥11,000
月度收益
节省人力（等效+4个开发人员）： ¥80,000
质量提升减少返工： ¥10,000
ROI：约8:1（每投入1元，产生8元价值）
结论：AI Agent不是成本，是投资。回报周期通常3个月内即可回本。
6.7 风险控制
核心理念：Agent不取代人
本章描述的所有自动化流程，都有一个前提：最终决定权在人。
4道风险防线
第1道：方案审核
Planner技术方案 → 架构师/技术负责人审核
方向错了，代码再好也没用
第2道：代码审查
Coder代码 → Reviewer Agent审查
Critical问题自动打回修复
第3道：人工兜底
所有Agent生成的PR必须经过人工审核才能合并
不可跳过，不可自动合并
第4道：回滚机制
每个PR附带回滚方案
保留Agent运行日志，便于问题追踪
必须避免的"自动化陷阱"
| 陷阱 | 表现 | 防范措施 |
|---|---|---|
| 过度信任 | 认为Agent生成的代码一定正确 | 强制人工审核，不可跳过 |
| Prompt退化 | Prompt越写越随意，质量下降 | 模板化、标准化Prompt |
| 工具依赖 | 离开Agent不会写代码 | 定期"无AI日"保持基本功 |
| 黑盒化 | 没人知道Agent生成的代码做了什么 | 保留Agent日志和决策链 |
| 范围蔓延 | Agent什么任务都接 | 明确Agent的职责边界 |
金仕达三句话铁律
1. Agent写的每一行代码，都有人看过
2. Agent做的每一个决策，都有人确认过
3. Agent出的每一个问题，都有人能解决
AI越强，越需要约束。这三句话不是限制Agent能力，而是确保Agent不成为组织的风险点。
本章小结
| 要点 | 内容 |
|---|---|
| 落地三步走 | 试点 → 推广 → 规模化，每阶段有明确门槛 |
| 需求转化 | 从业务语言提取结构化任务，拆解Agent任务树 |
| Planner方案 | AI自动生成结构化技术方案，作为"需求到代码"桥梁 |
| 代码生成 | Coder Agent基于方案生成代码，Reviewer Agent自动审查 |
| 自动PR | AutoPR引擎自动创建分支、提交代码、创建PR、触发CI/CD |
| 效果度量 | 效率提升5-6倍（金仕达实测数据），ROI约8:1 |
| 风险控制 | 4道防线：方案审核→代码审查→人工兜底→回滚机制 |
| 核心理念 | Agent辅助人，不取代人；最终决定权在开发者手中 |
思考与练习
- 诊断：盘点你所在组织的开发流程，识别最耗时的3个环节，评估哪些适合引入Agent
- 规划：如果要在你的团队推Agent落地，选哪个项目作为试点？理由是什么？
- 度量：设计你团队的Agent落地KPI体系，哪些指标最能反映真实效果？
- 风险：全面接入Agent后，你最担心的3个风险是什么？准备怎么应对？
- 实战：参考本章AutoPR代码，为你团队搭建一个简单的"需求到PR"自动化流水线
下一章：第7章：Prompt库与知识沉淀
第7章：Prompt库与知识沉淀
学习本章后，你将：建立系统的Prompt库，掌握Prompt版本管理，能够沉淀和复用Agent知识。
7.1 为什么需要Prompt库
经过前几章的学习，你已经能写出不错的Prompt。但你会发现一个常见问题：每次都在重复写"你是一个资深Go开发者"。这不仅是打字累的问题。
重复造轮子
每次跟AI交互都从零写Prompt，浪费时间和token。今天写得好的Prompt，明天想用找不到了。没有Prompt库，你每次都在从零开始。
团队无法复用
假设你的团队有5个开发者在用AI：
| 开发者 | Prompt水平 | 问题 |
|---|---|---|
| 小明 | 高 | 但他忙，没时间分享 |
| 小红 | 中 | 关键看状态 |
| 小李 | 低 | 不知道怎么改进 |
| 小张 | 中 | 每次写的不一样 |
| 你 | 高 | 但你的经验无法复制 |
团队最好的Prompt在小明脑子里——如果他离职了，经验也随之带走。Prompt库的作用就是把个人经验转化为团队资产。
无法追溯和迭代
没有版本记录，你不知道改了什么、哪个版本效果最好。
| 问题 | 没有Prompt库 | 有Prompt库 |
|---|---|---|
| 重复建设 | 每次从零写 | 直接复用模板 |
| 团队复用 | 经验在个人脑子里 | 经验在共享库里 |
| 版本追溯 | 不知道改了什么 | 完整变更历史 |
7.2 Prompt的分类体系
按场景分类
| 大类 | 场景 | 用途 | 典型角色 |
|---|---|---|---|
| 编码 | 代码生成 | 生成功能代码 | Go/Python/Java开发者 |
| 编码 | 代码审查 | 代码质量检查 | 高级工程师/架构师 |
| 编码 | 重构优化 | 代码重构、性能优化 | 重构专家 |
| 编码 | 单元测试 | 生成单测用例 | QA工程师 |
| 需求 | 需求分析 | 分析需求完整性 | 产品经理 |
| 设计 | 架构设计 | 系统设计评审 | 架构师 |
| 设计 | 接口设计 | API/协议设计 | 后端架构师 |
| 审查 | 方案审查 | 技术方案review | 技术主管 |
| 审查 | 安全审查 | 安全性检查 | 安全工程师 |
| 文档 | 技术文档 | API文档/设计文档 | 技术写手 |
| 运维 | 问题排查 | 故障分析诊断 | SRE |
| 运维 | 性能分析 | 性能瓶颈定位 | 性能工程师 |
按复杂度分层
| 层级 | 描述 | 示例 |
|---|---|---|
| L1 - 简单指令 | 一句话让AI做事 | "用Python写一个冒泡排序" |
| L2 - 结构模板 | 带分段和格式要求 | 角色+任务+约束三段式 |
| L3 - 复合Prompt | 多步骤+条件分支 | "先分析需求，再设计中立方案，最后评估风险" |
| L4 - Agent Prompt | 含工具调用和决策逻辑 | "你是一个编码Agent，当遇到安全问题时调用snyk_review工具" |
编码规则
格式：[大类]-[场景]-[序号]
示例：COD-REVIEW-001（Go代码审查模板）
REQ-ANALYSIS-001（需求完整性分析模板）
DES-API-003（RESTful API设计评审模板）
7.3 Prompt模板设计
每个Prompt模板应该像一段"函数代码"——有参数、有逻辑、有输出约定。
变量模板
用 {{variable}} 语法标记可替换部分。
你是一个 {{role}} 开发者，帮我完成以下任务：
任务描述：{{task_description}}
技术约束：
- 编程语言：{{language}}
- 框架版本：{{framework_version}}
- 性能要求：{{performance_requirement}}
输出要求：
- 格式：{{output_format}}
- 包含单元测试：{{include_tests}}
- 注释风格：{{comment_style}}
使用示例：
你是一个 Go 开发者，帮我完成以下任务：
任务描述：实现一个线程安全的连接池，支持超时控制和健康检查
技术约束：
- 编程语言：Go 1.22
- 框架版本：标准库
- 性能要求：并发1000连接，内存占用<50MB
输出要求：
- 格式：完整可编译的Go文件
- 包含单元测试：是
- 注释风格：Go标准注释
条件块
当模板需要支持多种角色变体时，使用条件语法。
你是一个 {{role}}。
{% if role == "架构师" %}
请评审：系统可扩展性、技术选型合理性、风险与降级方案
{% elif role == "安全工程师" %}
请评审：认证授权机制、数据传输加密、安全漏洞排查
{% elif role == "性能工程师" %}
请评审：响应时间与吞吐量、资源消耗、瓶颈点分析
{% endif %}
方案内容：
{{scheme_content}}
组合模式
将多个模板组合成工作流。
基础模板：COD-CODE-001（代码生成）
组合模板：COD-WORKFLOW-001（完整编码工作流）
步骤：
1. COD-CODE-001 → 生成功能代码
2. COD-REVIEW-001 → 代码审查
3. COD-UNITTEST-002 → 生成单元测试
4. 审查未通过则返回步骤1重试（最多3次）
元信息
每个模板都应有完整的元信息。
---
id: COD-CODE-001
name: Go代码生成标准模板
version: 2.1
author: 李冰成
created: 2025-06-01
updated: 2025-12-15
tags: [go, coding, standard]
avg_score: 4.5/5.0
usage_count: 328
---
7.4 Prompt版本管理
Prompt和代码一样，需要版本管理。最直接的方式：用Git管理Prompt。
目录结构
prompt-library/
├── README.md # 使用说明
├── CHANGELOG.md # 全局变更日志
├── categories/ # 分类索引
│ ├── coding.md
│ └── review.md
├── templates/ # Prompt模板文件
│ ├── coding/
│ │ ├── COD-CODE-001-go-gen.md
│ │ ├── COD-REVIEW-001-go-review.md
│ │ └── COD-UNITTEST-001-gen.md
│ ├── review/
│ │ ├── REV-SCHEME-001-arch-review.md
│ │ └── REV-SEC-001-security.md
│ ├── design/
│ │ ├── DES-ARCH-001-system-design.md
│ │ └── DES-API-001-restful.md
│ └── ops/
│ ├── OPS-DEBUG-001-troubleshoot.md
│ └── OPS-PERF-001-analysis.md
├── workflows/ # 组合工作流
├── evals/ # 评估记录
└── scripts/ # 辅助工具
版本号约定
格式：v主版本.次版本
v1.0 - 初始版本
v1.1 - 增加变量支持，修复边界情况
v2.0 - 重大重构，兼容性不保证
变更日志
# COD-CODE-001: Go代码生成模板
## v2.1 (2025-12-15)
### Changed
- 优化性能约束描述，从"性能要求高"改为具体量化指标
- 增加Go 1.22新特性提示
### Added
- 增加`{{comment_style}}`变量，支持三种注释风格
## v2.0 (2025-10-01)
### Changed
- 从L2结构模板升级为L3复合Prompt，增加条件块支持
## v1.0 (2025-06-01)
### Added
- 初始版本，支持基础代码生成需求
分支策略
| 分支 | 用途 |
|---|---|
| main | 稳定版本，经过评估审核 |
| staging | 预发布，待评估的新模板 |
| dev | 开发中，实验性模板 |
流程： dev开发 → 团队评审 → staging上线评估（1-2周） → 评估通过合并到main，不通过则回滚。
Prompt diff
v1.0: "你是一个Go开发者，帮我写代码"
v2.0: "你是一个资深Go开发者..."
+ "专注于后端服务开发"
+ "熟悉Go 1.22的新特性"
+ "代码风格遵循Uber Go Style Guide"
diff让你知道：改了哪里、为什么改、效果如何。
7.5 Prompt测试与评估
评估维度
| 维度 | 定义 | 评分（1-5） |
|---|---|---|
| 准确性 | AI输出是否符合Prompt要求 | 5：完全符合 |
| 完整性 | 是否覆盖所有要点 | 5：全部覆盖 |
| 一致性 | 多次执行结果是否稳定 | 5：稳定一致 |
| 效率 | 生成速度 + token消耗 | 5：又快又省 |
| 可用性 | 输出是否需要大量修改 | 5：可直接用 |
评估流程
评估周期：每个版本至少评估3次，取平均值
步骤：
1. 准备测试用例（3-5个典型场景）
2. 每个用例执行3次（temperature=0.3）
3. 按5个维度打分
4. 计算综合得分，记录到 evals/ 目录
A/B测试
当不确定两个版本哪个更好时，做A/B测试。
┌─────────────────────────────────────────────────────┐
│ A/B测试流程 │
├─────────────────────────────────────────────────────┤
│ │
│ 用户请求 │
│ │ │
│ ├── 50% ──► Prompt v1.0 ──► AI输出 ──► 记录结果 │
│ │ │
│ └── 50% ──► Prompt v2.0 ──► AI输出 ──► 记录结果 │
│ │
│ 对比： │
│ ┌──────────────┬───────┬───────┬────────┐ │
│ │ 指标 │ v1.0 │ v2.0 │ 差异 │ │
│ ├──────────────┼───────┼───────┼────────┤ │
│ │ 平均分 │ 4.2 │ 4.5 │ +7.1% │ │
│ │ 生成时间(ms) │ 3200 │ 2800 │ -12.5% │ │
│ │ 用户修改率 │ 35% │ 22% │ -37.1% │ │
│ └──────────────┴───────┴───────┴────────┘ │
│ │
│ 结论：v2.0全面优于v1.0，建议升级 │
│ │
└─────────────────────────────────────────────────────┘
评估记录示例
# 评估报告：COD-CODE-001 v2.1
评估日期：2025-12-20
评估人：李冰成
测试模型：Claude 3.5 Sonnet
| 用例 | 描述 | 准确性 | 完整性 | 一致性 | 效率 | 可用性 |
|---|---|---|---|---|---|---|
| 用例1 | HTTP服务基础框架 | 5 | 5 | 4 | 4 | 5 |
| 用例2 | 配置热加载实现 | 4 | 5 | 4 | 4 | 4 |
| 用例3 | 数据库迁移脚本 | 5 | 4 | 5 | 5 | 5 |
| 用例4 | gRPC服务端实现 | 4 | 4 | 4 | 3 | 4 |
综合评分：4.44（较v2.0提升+0.16）
结论：建议发布到main分支。
7.6 知识沉淀体系
三层知识结构
┌─────────────────────────────────────────────────────┐
│ 第一层：Patterns │
│ 成功的Prompt用法、有效的协作模式、已验证的方法论 │
├─────────────────────────────────────────────────────┤
│ 第二层：Anti-Patterns │
│ 失败的Prompt用法、没用的技巧、踩过的坑 │
├─────────────────────────────────────────────────────┤
│ 第三层：最佳实践文档 │
│ 完整的实践指南，包含多个Pattern和Anti-Pattern案例 │
└─────────────────────────────────────────────────────┘
Pattern记录模板
Pattern记录使用以下格式：
## Pattern: [名称]
### 解决的问题
[一句话描述]
### 示例Prompt
[完整的Prompt示例，缩进4个空格]
### 适用场景
- [场景1]
### 不适用场景
- [场景1]
### 效果指标
- 输出质量提升：[量化数据]
- 效率提升：[量化数据]
Anti-Pattern记录模板
Anti-Pattern记录使用以下格式：
## Anti-Pattern: [名称]
### 错误描述
[当时怎么做的，导致什么结果]
### 错误示例
[错误的Prompt示例，缩进4个空格]
### 正确做法
[正确的Prompt示例，缩进4个空格]
### 经验教训
[从中学到了什么]
知识库目录结构
knowledge-base/
├── patterns/ # 成功模式
│ ├── coding/
│ │ ├── pattern-go-context.md
│ │ └── pattern-error-handling.md
│ └── review/
│ └── pattern-security-first.md
├── anti-patterns/ # 反模式
│ ├── coding/
│ │ └── anti-goroutine-leak.md
│ └── review/
│ └── anti-superficial-review.md
└── best-practices/ # 最佳实践
├── go-concurrency-guide.md
└── code-review-checklist.md
持续沉淀机制
日度：随手记录好用的Prompt和AI表现不好的情况
周度：选1-2个有价值的整理为标准格式，纳入知识库
月度：分享会，每人分享一个Pattern或Anti-Pattern
季度：全面评估Prompt库使用情况，废弃/优化模板，发季度报告
经验沉淀黄金法则
| 类型 | 要不要记 | 原因 |
|---|---|---|
| 成功的Pattern | 必须记 | 让经验可复制 |
| 失败的尝试 | 必须记 | 让别人不踩同样的坑 |
| 他人的经验 | 推荐记 | 团队知识共享 |
| 踩过的坑 | 第一个记 | 教训比经验更值钱 |
本章小结
| 要点 | 内容 |
|---|---|
| 为什么需要Prompt库 | 避免重复造轮子、团队复用、版本追溯，把个人经验转为团队资产 |
| Prompt分类体系 | 按场景（编码/需求/设计/审查/文档/运维）、复杂度（L1-L4）分类，统一编码 |
| Prompt模板设计 | 变量替换（{{variable}}）、条件分支、组合模式，含元信息 |
| 版本管理 | Git管理，v主版本.次版本，CHANGELOG，分支策略保障质量 |
| 测试与评估 | 5维评估（准确性/完整性/一致性/效率/可用性），A/B测试驱动优化 |
| 知识沉淀体系 | Patterns+Anti-Patterns+Best Practices三层结构，定期沉淀机制 |
思考与练习
- 盘点：你现在使用的Prompt中，有多少是"每次从零写的"？如果建立Prompt库，能节省多少时间？
- 分类：按照7.2节的分类体系，把你常用的Prompt分类编码，至少列5个
- 模板化：选一个你每天都会用的Prompt，按7.3节的变量模板重新设计，对比效果
- 版本记录：为你的Prompt建立Git仓库，写CHANGELOG
- A/B测试：把一个Prompt改进前后做A/B测试，用7.5节的5维评估表打分对比
- 沉淀：回想你使用AI过程中踩过的坑，写一个Anti-Pattern记录
下一章：第8章：实战案例：金仕达AI编码助理
第8章：实战案例：某金融科技公司AI编码助理
学习本章后，你将：通过一个完整的企业级案例，理解Agent系统从0到1的全过程，掌握实际落地中的关键决策和技术选型。
8.1 项目背景
某金融科技公司是谁
上海某金融科技公司软件科技有限公司成立于1995年，是中国领先的金融科技服务商，长期服务于上海黄金交易所、工商银行、招商银行、平安银行等头部金融机构，业务覆盖贵金属、固收、量化交易等核心金融领域。公司研发团队规模在300～500人之间，技术栈以C++ + Java为主，同时广泛使用Go语言进行云平台和中间件开发。
为什么需要AI编码助理
2024年初，团队面临几个现实问题：
1. 研发效能瓶颈
金融科技项目的交付周期越来越短，但代码质量要求丝毫不能降低。一个典型的需求从开发到上线，经历需求分析、方案设计、编码实现、代码审查、测试验证、部署上线六个环节，每个环节都是人的瓶颈。
| 环节 | 耗时占比 | 瓶颈 |
|---|---|---|
| 需求分析 | 15% | 需求理解偏差导致返工 |
| 方案设计 | 10% | 设计质量依赖个人经验 |
| 编码实现 | 35% | 纯手工编码效率有限 |
| 代码审查 | 15% | 审查深度和时间矛盾 |
| 测试验证 | 20% | 测试覆盖率和执行速度 |
| 部署上线 | 5% | 自动化程度较高 |
编码和审查占据了50%的工作量，且严重依赖资深工程师。
2. 人员流动风险
金融科技行业的资深开发者流动率高。一个核心模块的开发者离职，往往意味着数月的生产力损失——新员工需要"爬代码"、理解业务逻辑、适应团队规范。
3. 技术债务积累
快速交付压力下，代码规范执行不到位、单测覆盖率低、文档更新滞后。技术债务像滚雪球，最终拖慢整个团队的交付速度。
核心诉求
作为一个云平台架构师和开发经理，我关注的是：能不能通过AI Agent，把团队的开发效能提升一个数量级，同时让知识经验可复制、可沉淀、可规模化？
调研后确定了两个关键判断：
- 通用AI编程工具（如GitHub Copilot）只能辅助"写代码"这个环节，无法覆盖从需求到PR的完整链路
- 金融行业对代码安全和合规有特殊要求，通用工具无法满足
这引出一个结论：我们需要一个面向某金融科技公司研发团队的AI编码助理，基于我们自己积累的技术规范和业务知识，覆盖从需求到PR的全流程。
8.2 需求定义
用户是谁
| 用户角色 | 人数 | 核心需求 |
|---|---|---|
| 一线开发者 | 30-40人 | 减少重复编码，提高开发效率 |
| 代码审查者 | 5-8人 | 自动化部分审查工作，聚焦逻辑问题 |
| 技术经理 | 3-5人 | 把控代码质量，跟踪交付进度 |
| 架构师 | 2-3人 | 确保方案设计符合技术规范 |
解决什么问题
核心价值：研发效能提升5-6倍。
这个目标的分解：
| 维度 | 当前现状 | 目标状态 |
|---|---|---|
| 编码效率 | 一个CRUD接口需30分钟 | 5分钟生成+5分钟审查 |
| 代码审查 | 人工审查，平均2小时/次 | AI初审+人工终审，30分钟 |
| 单测覆盖 | 覆盖率不足40%，手动编写 | AI自动生成，目标80%+ |
| 文档同步 | 滞后，经常忘记更新 | AI自动生成更新 |
| 新人上手 | 2-3周才能独立开发 | 1周内能提交代码 |
功能需求优先级
| 优先级 | 功能 | 说明 |
|---|---|---|
| P0 | 需求解析 | 将自然语言需求转为结构化任务 |
| P0 | 代码生成 | 根据需求生成符合规范的代码 |
| P0 | 代码审查 | 自动化代码审查，拦截常见问题 |
| P1 | 方案生成 | 自动生成技术方案设计 |
| P1 | PR自动提交 | 生成代码后自动创建PR |
| P2 | 单测生成 | 自动生成单元测试 |
| P2 | 文档生成 | 自动生成接口文档 |
非功能性需求
| 需求 | 指标 | 原因 |
|---|---|---|
| 响应时间 | 单次任务<30秒 | 不能打断开发者的心流 |
| 代码安全 | 代码不出内网 | 金融合规要求 |
| 合规审查 | 所有生成代码需记录 | 审计追踪 |
| 定制能力 | Skill可自定义 | 适配不同项目规范 |
| 离线容错 | 网络中断可降级 | 内网环境稳定性 |
8.3 技术选型
为什么不是LangChain
在技术选型阶段，我们评估了三种方案：
| 维度 | LangChain | AutoGPT | 自研Agent框架 |
|---|---|---|---|
| 灵活性 | 中，受框架约束 | 低，流程固化 | 高，完全可控 |
| 定制能力 | 中，需写复杂Chain | 低，预设Agent | 高，可定制每个Skill |
| 内网部署 | 难，依赖云LLM API | 难 | 易，可对接内网模型 |
| 金融合规 | 不满足 | 不满足 | 可定制审计 |
| 学习成本 | 高，API复杂 | 中 | 取决于框架设计 |
| 社区生态 | 丰富 | 一般 | 自建 |
| 性能开销 | 高，Python重 | 高 | 可优化 |
| Skill复用 | 通过Tool/Chain | 不适用 | Skill定义即复用 |
最终选择自研的原因：
- 金融场景特殊性：某金融科技公司的代码需要遵循特定的业务规范和技术约束，通用框架无法内置这些规则
- 合规要求：所有AI生成代码必须可审计、可追溯，LangChain的Callbacks机制无法满足金融级审计需求
- 内网部署：开发环境在内网，不能直接调用公网LLM API，需要对接私有化部署的模型
- 性能需求：开发者不能等，30秒内必须出结果，需要精细的流式处理和缓存优化
技术架构选型
| 决策 | 选择 | 理由 |
|---|---|---|
| 编程语言 | Go | 团队Go能力强，性能好，适合构建Agent Runtime |
| LLM模型 | 通义千问+GLM | 国产模型，支持私有化部署 |
| 消息队列 | NSQ | 已经在生产环境使用，运维经验丰富 |
| 存储 | MySQL + Redis | 现有基础设施，无需引入新中间件 |
| 代码平台 | GitHub | 团队已在用，API成熟 |
| 运行时 | Docker | 内网容器化部署，方便扩展 |
AIWork底座的关系
AI编码助理不是从零开始的——它构建在团队已有的AIWork底座之上。AIWork提供了Skill-Engine、Context Management、LLM Gateway等基础能力，而AI编码助理是在其之上构建的一套面向编码场景的Skill集合。
┌──────────────────────────────────────┐
│ AI编码助理（面向场景的Skill层） │
│ 需求解析Skill 代码生成Skill 审查Skill │
│ 方案生成Skill PR提交Skill 测试Skill │
└──────────────────┬───────────────────┘
│ 调用
┌──────────────────▼───────────────────┐
│ AIWork 底座（基础设施） │
│ Skill-Engine │ Context Manager │
│ LLM Gateway │ Prompt 库 │
│ 规则引擎 │ 审计日志 │
└──────────────────────────────────────┘
8.4 架构设计
整体架构
┌─────────────────────────────┐
│ 用户界面 │
│ GitHub PR / Web Dashboard │
└─────────────┬───────────────┘
│
┌─────────────▼───────────────┐
│ API Gateway │
│ 认证(RBAC) 限流 路由 │
└─────────────┬───────────────┘
│
┌─────────────▼───────────────┐
│ Agent Runtime │
│ 任务分解 调度 状态管理 │
└─────────────┬───────────────┘
│
┌─────────────────────────┼─────────────────────────┐
│ │ │
┌─────▼──────┐ ┌──────▼──────┐ ┌───────▼─────┐
│ Skill │ │ Skill │ │ Skill │
│ Engine │ │ Engine │ │ Engine │
│ (需求解析) │ │ (代码生成) │ │ (代码审查) │
└─────┬──────┘ └──────┬──────┘ └───────┬─────┘
│ │ │
└────────────────────────┼─────────────────────────┘
│
┌──────────────▼──────────────┐
│ LLM Gateway │
│ 通义千问 GLM 统一API │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ 外部系统 │
│ GitHub API 内网GitLab │
└─────────────────────────────┘
核心模块
1. LLM Gateway
所有模型调用的统一入口，支持：
- 多模型路由（通义千问处理代码生成，GLM处理审查逻辑）
- 请求重试和降级（一个模型超时自动切换）
- Token用量统计和成本控制
- 审计日志记录每次调用的输入输出
LLM Gateway 请求流程：
请求 → 认证 → 路由选择 → 模型调用 → 结果校验 → 审计记录 → 返回
│
┌───────┴───────┐
│ 路由策略 │
├───────────────┤
│ 代码类 → 通义 │
│ 审查类 → GLM │
│ 文档类 → 通义 │
└───────────────┘
2. Skill Engine
在AIWork的Skill-Engine基础上，定制了编码场景专用的执行流程：
Skill 执行流程（以代码生成为例）：
输入: {"requirement": "实现用户登录接口", "project": "trading-api"}
│
▼
┌────────────────────────────┐
│ Step 1: 需求解析 │
│ 解析自然语言需求 │
│ 提取功能点、接口、参数 │
│ 输出结构化需求文档 │
└────────────────────────────┘
│
▼
┌────────────────────────────┐
│ Step 2: 上下文加载 │
│ 加载项目代码规范（规则引擎） │
│ 加载相关代码库（RAG） │
│ 加载Git历史记录 │
└────────────────────────────┘
│
▼
┌────────────────────────────┐
│ Step 3: Prompt组装 │
│ 角色设定：金融科技开发者 │
│ 注入规范：命名规则、异常处理 │
│ 注入参考：相似代码片段 │
└────────────────────────────┘
│
▼
┌────────────────────────────┐
│ Step 4: LLM调用 │
│ 调用通义千问 │
│ 流式输出（边生成边返回） │
└────────────────────────────┘
│
▼
┌────────────────────────────┐
│ Step 5: 结果后处理 │
│ 格式校验 │
│ 安全检查（敏感信息泄露） │
│ 自动格式化（gofmt/javafmt）│
└────────────────────────────┘
│
▼
输出: {"files": [...], "summary": "..."}
3. 规则引擎
某金融科技公司多年的开发规范和技术积累，全部沉淀为可执行的规则：
规则引擎架构：
┌─────────────────────────────────────────────┐
│ 规则引擎 │
├─────────────────────────────────────────────┤
│ ┌─────────────┐ ┌──────────────────────┐ │
│ │ 代码规范规则 │ │ 业务逻辑规则 │ │
│ │ - 命名规范 │ │ - 交易接口协议 │ │
│ │ - 包结构规范 │ │ - 错误码定义 │ │
│ │ - 注释规范 │ │ - 日志规范 │ │
│ │ - 异常处理 │ │ - 审计字段要求 │ │
│ └─────────────┘ └──────────────────────┘ │
│ ┌─────────────┐ ┌──────────────────────┐ │
│ │ 安全规则 │ │ 架构规则 │ │
│ │ - SQL注入 │ │ - 分层依赖方向 │ │
│ │ - XSS │ │ - 接口设计原则 │ │
│ │ - 敏感信息 │ │ - 事务边界控制 │ │
│ │ - 权限校验 │ │ - 缓存策略 │ │
│ └─────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────┘
4. Prompt库
针对编码场景设计了专用Prompt模板库，每个Skill对应一个精心设计的模板：
# 代码生成Skill的Prompt模板（简化版）
name: code_generation
version: 2.1.0
category: coding
prompt_template:
template: |
你是一个精通{language}的金融科技开发者。请根据以下需求生成代码。
## 项目上下文
- 项目：{project_name}
- 模块：{module_name}
- 使用的框架：{framework}
## 代码规范要点
{code_rules}
## 参考代码片段
{reference_code}
## 需求描述
{requirement}
## 输出要求
1. 生成完整的可编译代码文件
2. 每行代码都要添加必要的注释
3. 包含完整的异常处理和日志记录
4. 遵循项目已有的包结构和命名规范
5. 不要生成测试代码（测试由单独的Skill处理）
## 输出格式
每个文件用以下标记分隔：
=== FILE: 文件路径 ===
文件内容
input_variables:
- name: language
type: string
required: true
- name: project_name
type: string
required: true
- name: module_name
type: string
required: true
- name: framework
type: string
required: true
- name: code_rules
type: string
required: true
- name: reference_code
type: string
required: false
- name: requirement
type: string
required: true
8.5 核心功能实现
8.5.1 需求解析Skill
将自然语言需求转化为结构化的开发任务。
执行流程：
输入："实现用户登录接口，支持用户名密码登录和手机验证码登录"
↓
输出：
{
"title": "用户登录接口实现",
"endpoints": [
{
"method": "POST",
"path": "/api/v1/login/password",
"params": {"username": "string", "password": "string"},
"response": {"token": "string", "expires_in": "int"}
},
{
"method": "POST",
"path": "/api/v1/login/sms",
"params": {"phone": "string", "code": "string"},
"response": {"token": "string", "expires_in": "int"}
}
],
"affected_modules": ["auth-service", "user-service"],
"estimated_effort": "4h",
"risks": ["短信发送依赖外部服务，需考虑降级"],
"test_cases": [
"用户名密码正确返回token",
"密码错误返回401",
"手机验证码过期返回错误"
]
}
核心代码片段（Go）：
// skill/requirement_parser.go
type RequirementParser struct {
llm *LLMGateway
schema *RequirementSchema
}
type RequirementInput struct {
Text string `json:"text"`
Project string `json:"project"`
Module string `json:"module"`
}
type RequirementOutput struct {
Title string `json:"title"`
Endpoints []Endpoint `json:"endpoints"`
AffectedMods []string `json:"affected_modules"`
Effort string `json:"estimated_effort"`
Risks []string `json:"risks"`
TestCases []string `json:"test_cases"`
}
func (p *RequirementParser) Execute(ctx context.Context, input RequirementInput) (*RequirementOutput, error) {
// 1. 加载项目的接口规范
rules := p.loadProjectRules(input.Project)
// 2. 组装Prompt
prompt := p.buildPrompt(input, rules)
// 3. 调用LLM（结构化输出）
result, err := p.llm.Chat(ctx, &LLMRequest{
Prompt: prompt,
OutputSchema: p.schema,
Temperature: 0.1, // 低温度，保证输出稳定
})
// 4. 解析和验证
output, err := p.parseResult(result)
if err != nil {
return nil, fmt.Errorf("解析需求失败: %w", err)
}
// 5. 记录审计日志
p.audit.Log(ctx, &AuditEntry{
Action: "requirement_parse",
Input: input.Text,
Output: output,
User: ctx.Value("user_id"),
})
return output, nil
}
8.5.2 方案生成Skill
在编码之前，先生成技术方案，由开发者确认后再进入编码阶段。
输入（来自需求解析Skill的结构化需求）：
- 功能：用户登录接口
- 接口数：2个
- 涉及模块：auth-service, user-service
- 风险：短信服务降级
↓
输出（技术方案）：
┌─────────────────────────────────────────────┐
│ 技术方案：用户登录接口 │
├─────────────────────────────────────────────┤
│ │
│ 1. 方案概述 │
│ 新增2个登录接口，复用现有auth-service │
│ │
│ 2. 接口设计 │
│ POST /api/v1/login/password │
│ POST /api/v1/login/sms │
│ │
│ 3. 数据流 │
│ 请求 → auth-service → user-service验证 │
│ ↓ │
│ 生成JWT Token │
│ ↓ │
│ 返回Token │
│ │
│ 4. 异常处理 │
│ 密码错误 → 返回401 + 错误码 │
│ 验证码过期 → 返回429 + 重试提示 │
│ 短信服务故障 → 降级为仅密码登录 │
│ │
│ 5. 安全考虑 │
│ 登录失败不提示具体原因（防枚举） │
│ 添加登录频率限制（5次/分钟/IP） │
│ 密码传输使用HTTPS + 前端加密 │
│ │
│ 6. 测试要点 │
│ - 正常流程：正确凭据返回Token │
│ - 异常流程：各失败场景 │
│ - 安全测试：暴力破解防护 │
│ - 压力测试：100并发无失败 │
│ │
└─────────────────────────────────────────────┘
为什么需要方案生成？ 从实践来看，让AI直接生成代码而不经过方案设计，容易出现偏离架构设计的情况。方案生成环节让开发者"先确认再编码"，相当于在AI产出的基础上增加了一层人工校验。
8.5.3 代码生成Skill
这是整个系统最核心的Skill。
Prompt设计的几个关键决策：
1. 角色注入
你是一个精通Java和SpringCloud的金融科技开发者，在某金融科技公司工作5年，
熟悉贵金属交易系统和银行接口协议。
不是简单地说"你是一个开发者"，而是给出具体的场景和年限，这能让LLM的输出风格更贴近某金融科技公司的实际代码风格。
2. 规则注入
把某金融科技公司的代码规范作为上下文注入Prompt：
code_rules: |
1. 所有Controller统一使用 @RestController，不使用 @Controller
2. Service层必须面向接口编程，Impl后缀实现类
3. 异常统一使用 BizException，禁止直接 throw RuntimeException
4. 日志使用 SLF4j，必须包含 traceId 用于链路追踪
5. 所有外部接口调用必须添加 @HystrixCommand 熔断
6. SQL语句必须使用 MyBatis-Plus，禁止手写JDBC
3. 参考代码注入
通过RAG检索项目代码库中相似功能的已有代码，作为Few-shot示例提供给LLM。这一步对代码质量提升最为显著。
代码生成质量对比（内部测试）：
| 维度 | 无参考代码 | 有参考代码 | 提升 |
|------|-----------|-----------|------|
| 格式合规率 | 67% | 94% | +27% |
| 直接可用率 | 42% | 78% | +36% |
| 平均修改行数 | 35行 | 12行 | -66% |
流式输出实现
开发者的等待体验至关重要。我们实现了基于SSE的流式输出：
// handler/stream_handler.go
func (h *StreamHandler) GenerateCode(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "Streaming not supported", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
req := parseRequest(r)
stream := h.codeGenSkill.ExecuteStream(r.Context(), req)
for chunk := range stream {
// 每生成一个代码块就推送给前端
data, _ := json.Marshal(chunk)
fmt.Fprintf(w, "data: %s\n\n", data)
flusher.Flush()
}
fmt.Fprintf(w, "data: {\"type\": \"done\"}\n\n")
flusher.Flush()
}
8.5.4 代码审查Skill
代码审查Skill是质量把控的关键防线。它独立于代码生成Skill运行——即使代码不是AI生成的，也可以提交审查。
审查维度：
| 审查维度 | 权重 | 检查项示例 |
|---|---|---|
| 功能正确性 | 30% | 逻辑完整、边界处理、状态流转 |
| 代码规范 | 20% | 命名、格式、包结构、注释 |
| 安全漏洞 | 25% | SQL注入、越权、敏感信息泄露 |
| 性能问题 | 15% | N+1查询、内存泄漏、锁竞争 |
| 可维护性 | 10% | 复杂度、重复代码、测试覆盖 |
审查结果格式：
{
"overall_score": 85,
"summary": "整体质量良好，存在2个中等风险和3个建议项",
"issues": [
{
"severity": "error",
"file": "AuthController.java",
"line": 45,
"type": "security",
"message": "登录接口未添加频率限制",
"suggestion": "添加 @RateLimiter 注解，限制每分钟5次请求"
},
{
"severity": "warning",
"file": "UserServiceImpl.java",
"line": 123,
"type": "performance",
"message": "循环中调用数据库查询，存在N+1问题",
"suggestion": "使用 IN 查询替代循环查询"
},
{
"severity": "suggestion",
"file": "LoginController.java",
"line": 10,
"type": "style",
"message": "缺少 @ApiOperation 注解",
"suggestion": "添加Swagger接口描述"
}
],
"approved": false,
"estimated_fix_time": "1h"
}
审查策略配置：
不同级别的代码，审查严格程度不同：
# review-policies.yaml
policies:
- name: critical-module
match: ["trade-engine", "settlement", "payment"]
required_approvers: 2
min_score: 90
check_security: true
check_performance: true
- name: normal-module
match: ["*"]
required_approvers: 1
min_score: 70
check_security: true
check_performance: false
- name: auto-approve
match: ["test/*", "config/*"]
min_score: 60
auto_approve: true
8.5.5 PR自动提交
当开发者确认生成的代码无误后，系统自动完成以下工作：
- 创建Git分支（基于master最新版本）
- 提交代码到分支
- 运行代码审查Skill
- 审查通过则自动创建PR
- PR描述自动填充（包含需求描述、变更文件、审查结果）
PR自动生成示例：
## 变更说明
实现用户登录接口（密码登录 + 短信验证码登录）
## 涉及文件
- auth-service/src/main/java/com/ks/controller/AuthController.java (新增)
- auth-service/src/main/java/com/ks/service/AuthService.java (新增)
- auth-service/src/main/java/com/ks/service/impl/AuthServiceImpl.java (新增)
## 接口定义
- POST /api/v1/login/password
- POST /api/v1/login/sms
## AI审查结果
- 评分：85/100
- 严重问题：0
- 警告：1（登录频率限制已添加，确认配置正确）
- 建议：2（补充Swagger注释，已处理）
## 自检清单
- [x] 代码符合项目规范
- [x] 异常处理完整
- [x] 日志记录完整
- [x] 安全审查通过
- [ ] 单元测试（AI生成中，见关联PR #142）
流程集成到GitHub Actions：
# .github/workflows/ai-pr.yml
name: AI Code Review
on:
pull_request:
types: [opened, synchronize]
jobs:
ai-review:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: AI Code Review
uses: ks-ai/code-review-action@v1
with:
api-key: ${{ secrets.AI_API_KEY }}
review-depth: full
check-security: true
min-score: 70
- name: Auto-approve if score >= 90
if: steps.review.outputs.score >= 90
uses: hmarr/auto-approve-action@v3
8.6 落地过程
推广策略
AI编码助理的推广分三个阶段：
第一阶段：试点团队（第1-2周）
| 步骤 | 内容 | 目标 |
|---|---|---|
| 选择试点 | 一个5-8人的后端开发小组 | 验证可行性 |
| 配置环境 | 部署内网LLM、配置项目规范 | 技术准备 |
| 培训 | 2小时Prompt工程师培训 | 让团队会用 |
| 试运行 | 从简单的CRUD接口开始 | 建立信心 |
试点团队选择标准：
- 中等规模项目（不是最核心的交易系统，风险可控）
- 开发者对AI接受度高
- 项目经理愿意配合
第二阶段：部门推广（第3-6周）
基于试点团队的反馈，完善系统后推向整个开发部门：
试点反馈 → 问题修复 → 功能增强 → 文档更新 → 部门培训 → 全面推广
关键改进（基于试点反馈）：
1. 流式输出让等待感降低80%
2. 加入参考代码后，代码可用率提升36%
3. 审查阈值从80分降到70分（初期太严格）
第三阶段：全公司推广（第7周后）
推广到其他技术团队，建立统一的AI开发标准。
培训体系
培训分为三个层级：
层级1：Prompt工程师培训
培训大纲（2小时）：
1. AI编码助理是什么（15分钟）
2. 怎么写好的需求Prompt（30分钟）
- 给的细节越多，结果越好
- 明确接口、参数、约束
- 举例说明
3. 常见场景演示（45分钟）
- 生成CRUD接口
- 生成单元测试
- 代码审查
4. 实操练习（30分钟）
- 每人完成一个真实需求
层级2：Skill编写规范
Skill编写培训（1小时）：
1. Skill是什么（10分钟）
2. Skill定义规范（20分钟）
- 命名规范
- 输入输出Schema设计
- Prompt模板编写要点
3. 如何测试一个Skill（15分钟）
4. 提交Skill的流程（15分钟）
层级3：最佳实践沉淀
团队每周分享AI辅助开发的最佳实践，形成知识库：
周最佳实践分享示例：
主题：如何让AI生成更符合项目规范的代码
发现：直接在Prompt中说"遵循项目规范"效果不好
改进：把规范的具体条款写进Prompt
效果：格式合规率从67%提升到94%
具体做法：
在Prompt中加入以下内容：
"本项目代码规范：..."
收集反馈
建立反馈闭环：
反馈收集机制：
1. PR评论：每次AI生成的PR添加标签，开发者可评价
2. 周问卷：每周5个问题，量化满意度
3. 月度复盘：回顾使用数据，讨论改进点
量化指标：
- 采纳率：AI生成的代码被保留的比例
- 满意度：开发者对AI输出的评分（1-5分）
- 问题率：AI生成的代码出Bug的比例
- 节省时间：每个需求节省的开发时间
8.7 效果数据
效能提升数据
经过3个月的运行，AI编码助理取得了以下效果：
| 指标 | 使用前 | 使用后 | 提升 |
|---|---|---|---|
| CRUD接口开发 | 30分钟/个 | 8分钟/个 | 3.75倍 |
| 代码审查 | 2小时/次 | 25分钟/次 | 4.8倍 |
| 单测覆盖率 | 38% | 82% | 2.16倍 |
| 新人上手时间 | 2.5周 | 1周 | 2.5倍 |
| 代码规范合规率 | 73% | 96% | +23% |
| Bug率（线上） | 3.2% | 1.1% | -66% |
"5-6倍效能提升"的完整含义：
这不是指"所有环节都提升5-6倍"，而是指一个完整需求的端到端交付周期：
一个典型需求（实现一个包含增删改查的后台管理接口）：
传统方式（分钟）：
需求理解 15 + 编码 30 + 自测 15 + 审查 120 + 修复 30 + 测试 60 = 270min
AI辅助方式（分钟）：
AI解析需求 5 + 确认方案 10 + AI生成代码 5 + 人工审查 25 + 修复 10 + 测试 20 = 75min
效率提升：270 / 75 = 3.6倍
加上AI生成的单测（节省60min）和自动化PR（节省15min）：
整体效率提升：(270 + 60 + 15) / 75 = 4.6倍
如果需求复杂度更高（多接口、多模块），倍数更大。
高频场景下（CRUD为主），实测可达5-6倍。
成本分析
| 成本项 | 月投入 | 说明 |
|---|---|---|
| LLM API调用 | 约5000元 | 内网部署，主要是GPU资源 |
| 服务器 | 约2000元 | Docker集群，4台服务器 |
| 维护人力 | 1人·周/月 | 规则更新、问题排查 |
| 培训成本 | 初期2000元 | 培训材料+时间成本 |
ROI计算：
团队规模：40人
平均薪资：25000元/月
AI带来的效能提升：按保守3倍计算
节省人力等效：40 × (1 - 1/3) ≈ 27人
月节省成本：27 × 25000 = 675000元
月投入成本：5000 + 2000 + 2500 ≈ 9500元
ROI = (675000 - 9500) / 9500 = 70倍
注意：这不是说"可以裁掉27个人"——实际上，释放出来的生产力被投入到更多有价值的工作中：代码重构、技术债务清理、技术创新。
用户反馈
开发者反馈抽样：
| 角色 | 反馈 |
|---|---|
| 高级开发 | "审查Skill帮我发现了几个我自己的盲点，现在每次提交前我都先让AI过一遍" |
| 中级开发 | "以前写个查询接口要翻半天文档，现在直接告诉AI需求就行了" |
| 新手开发 | "入职第一周就能提交PR了，AI生成的代码就是最好的学习范例" |
| 技术经理 | "代码规范已经不是问题了，AI生成的代码比大部分人写的都规范" |
不满意反馈：
| 问题 | 频率 | 改进措施 |
|---|---|---|
| 生成的代码有时不符合预期 | 15% | 加强需求解析阶段的确认 |
| 复杂业务逻辑处理不好 | 8% | 增加领域知识的RAG检索 |
| 审查结果误报 | 5% | 调整审查规则的阈值 |
8.8 经验总结
踩过的坑
坑1：第一版Prompt太通用
最开始，我们直接用了通用的代码生成Prompt，没有注入某金融科技公司的规范。结果生成的代码虽然语法正确，但风格天差地别——用Stream API的、用Lambda的、各种风格的都有，审查环节反而更累了。
教训：Prompt必须注入团队规范，标准越具体，AI输出越符合预期。
坑2：让AI一步生成所有代码
早期版本让AI一次性生成"全部代码"，结果输出质量不稳定——文件一多，LLM就开始"遗忘"前面的要求。
教训：一个Skill只做一件事。代码生成Skill只负责"生成一个文件"，而不是"生成整个项目"。多个文件通过编排组合。
坑3：审查标准一开始设太高
初版审查Skill的通过分数设在了90分，结果一周内通过率不到30%，开发者失去了使用耐心。
教训：从松到严。先让开发者"用起来"，再逐步收紧质量标准。初期设70分通过线，随着AI能力提升逐步调高。
坑4：忽略了"人"的因素
系统上线后我们发现：不是所有开发者都愿意用。有些资深开发者觉得"AI写的代码不够好"，有些开发者担心"用了AI体现不出自己的价值"。
教训：技术落地是人的问题。需要：
- 让"会用AI"成为团队的荣誉，而不是威胁
- 给早期使用者展示成果
- 把AI定位为"辅助工具"而非"替代品"
坑5：没有预留人工审核环节
早期版本的流程是：需求输入 → AI生成代码 → 自动PR。跳过人工审核，结果出了几次"看起来对但逻辑不对"的问题。
教训：AI编码助理是"辅助"不是"替代"。AI生成 + 人工审查 + AI二次验证 = 质量最稳定的模式。
关键决策回顾
| 决策 | 选项 | 选择 | 事后评价 |
|---|---|---|---|
| 自研 vs. 买 | 自研/LangChain | 自研 | 正确。框架满足不了金融合规 |
| 内网 vs. 公网 | 内网部署/调用云端 | 内网 | 正确。安全合规是第一优先级 |
| 流式 vs. 批量 | 流式输出/一次性返回 | 流式 | 正确。用户体验提升明显 |
| 规则引擎 vs. 纯Prompt | 规则+Prompt/纯Prompt | 规则+Prompt | 正确。规则比Prompt更稳定 |
| P0范围 | 功能范围大小 | 聚焦代码生成+审查 | 正确。先做好两个核心场景 |
给同行的建议
1. 从最痛的点切入
不是所有场景都适合AI。找到团队最痛、耗时最多、重复性最高的环节。在我们的场景里，编码和审查占了50%的工作量，是最值得优化的环节。
2. 先让人"用起来"，再追求完美
第一版不需要做到100分。能让开发者节省20%的时间，他们就会主动用。有了用户和数据，才有迭代的方向。
3. 技术规范是AI助手的"灵魂"
同样的LLM，不同的Prompt输出天差地别。团队的技术规范、代码风格、业务知识，是AI助手价值的核心来源。花时间沉淀规范，比花时间调模型参数更有价值。
4. 人是落地的最大变量
75%的落地失败是"人"的问题，不是"技术"的问题。培训、沟通、激励机制，和架构设计一样重要。
5. 建立反馈闭环
没有反馈就没有改进。每个AI生成的PR都标记是否被采纳、需要多少修改。这些数据是持续优化AI的基础。
未来规划
| 阶段 | 计划 | 目标 |
|---|---|---|
| 短期 | 支持更多项目配置 | 覆盖全部后端项目 |
| 中期 | 引入RAG检索业务文档 | AI能理解200+业务术语 |
| 长期 | Multi-Agent协作 | 需求评审→方案设计→编码→审查→测试全链路自动化 |
本章小结
| 维度 | 内容 |
|---|---|
| 项目背景 | 某金融科技公司是1995年成立的金融科技公司，30-60人研发团队面临效能瓶颈、人员流动、技术债务三大挑战 |
| 需求定义 | 面向一线开发者和审查者，核心目标是研发效能提升5-6倍，聚焦从需求到PR的完整链路 |
| 技术选型 | 自研Agent框架而非LangChain，核心原因是金融合规、内网部署、定制化需求无法满足 |
| 架构设计 | AIWork底座 + 面向编码场景的Skill层，LLM Gateway统一管理模型调用 |
| 核心功能 | 需求解析、方案生成、代码生成、代码审查、PR自动提交，5个Skill覆盖开发全流程 |
| 落地过程 | 三阶段推广：试点团队→部门推广→全公司，配合培训体系和反馈闭环 |
| 效果数据 | CRUD效率3.75倍、审查效率4.8倍、单测覆盖率从38%到82%、Bug率下降66% |
| 经验总结 | 从最痛点切入、先用人再完美、技术规范是灵魂、人是最大变量、建立反馈闭环 |
思考与练习
-
分析：你所在团队/公司的研发流程中，最耗时的三个环节是什么？哪个环节最适合AI辅助？
-
设计：假设你要在团队推广AI编码助理，你会选择哪个项目作为试点？选择标准是什么？
-
对比：本章介绍了LangChain和自研框架的对比。重新评估一下——如果你是CTO，面对一个初创团队（10人以下），你会怎么选？如果是一个500人的金融公司呢？
-
实践：如果你想在团队中引入AI代码审查，第一步应该做什么？（提示：想想本章说的"先让人用起来"）
-
反思：如果AI编码助理生成的PR通过率达到了95%，团队还需要人工代码审查吗？质量和效率的平衡点在哪里？
-
计算：用本章的ROI模型，估算一下在你的团队中引入AI编码助理的投入产出比。
** 本章节纯属虚构，如有雷同纯属巧合！！**
全书完结。感谢阅读！