Agent与工作流实战:搭建可落地的AI应用
你已经能用提示词让AI回答问题,也学会了用RAG让AI”知道”你的业务知识。但真实的生产环境不是一问一答——你需要AI能主动思考、调用工具、记住上下文、协调多个步骤。这就是Agent。
你会学到什么
学完这门课,你将能够:
- 理解Agent、工作流、Copilot的本质区别,知道什么时候用什么
- 用Function Calling让AI调用真实的外部API和工具
- 用ReAct模式搭建能”边想边做”的智能Agent
- 为Agent设计短期和长期记忆系统
- 根据业务需求选择Coze、Dify、n8n等工作流工具
- 实现一个Mini-Manus——能浏览网页、读写文件的自主Agent
- 搭建多Agent协作系统,让多个专精Agent分工配合
- 独立完成一个从需求分析到上线的AI应用项目
学习路径
第1章 理解Agent → 第2章 让AI调用工具 → 第3章 让AI边想边做
→ 第4章 给Agent加记忆 → 第5章 选工作流工具
→ 第6-7章 实战项目 → 第8章 完整应用交付
第1章 Agent:AI应用的下一个形态
本章目标:理解Agent为什么出现,以及它和ChatBot、工作流、Copilot的本质区别。
1.1 为什么ChatBot不够用了
传统ChatBot的工作模式:
用户提问 → AI回答 → 结束
这在简单问答场景够用,但面对真实业务有三个致命问题:
问题一:AI只能说话,不能做事。
你问”帮我查一下明天上海的天气”,AI只能说”我无法实时查询”——它没有”手”,不能主动获取信息。
问题二:AI没有记忆。
你昨天告诉它”我在上海工作”,今天再问”明天出门带不带伞”,它已经忘了你在哪个城市。
问题三:AI只能做一步。
真实需求往往是多步任务:“帮我调研三个竞品的价格,做个对比表,发到我邮箱。“这需要搜索→整理→制表→发邮件,是一个多步骤、多工具协作的复杂流程。
这三个问题催生了一个新方向:让AI从”只会回答”进化到”能完成任务”——这就是Agent。
1.2 Agent的出现原因和发展脉络
Agent的核心思想:AI不只是回答问题的工具,而是能感知环境、做出决策、调用工具、执行动作,并在此过程中维护状态的自主系统。
发展脉络:
| 阶段 | 大致时间 | 特征 | 代表产品 |
|---|---|---|---|
| 规则引擎 | 2015年前 | if-else硬编码,完全无智能 | 传统客服机器人 |
| ChatBot | 2022年起 | 基于大模型对话,单轮问答 | ChatGPT初期版本 |
| Copilot | 2023年起 | AI辅助人工作,人做决策 | GitHub Copilot |
| Agent | 2024年起 | AI主导执行,人做监督 | Manus、Devin |
| Multi-Agent | 2025年起 | 多个专精Agent协作 | Claude Code Agent、CrewAI |
用一个类比理解这个演进:
- 规则引擎 = 自动售货机(按按钮出货,固定逻辑)
- ChatBot = 电话客服(能回答问题,但不能帮你操作)
- Copilot = 副驾驶(你开车,他给你建议)
- Agent = 代驾(你说目的地,他自己开车去)
- Multi-Agent = 整个出租车公司(调度、驾驶、客服协作)
1.3 Agent vs 工作流 vs Copilot的区别
很多人混淆这三个概念。用一张表说清楚:
| 维度 | Copilot | 工作流 | Agent |
|---|---|---|---|
| 谁主导 | 人 | 预设流程 | AI |
| 决策方式 | 人做所有决策 | 按固定规则走 | AI自主判断 |
| 灵活性 | 高(人随时改) | 低(改流程需重新设计) | 中(有边界但能适应) |
| 可预测性 | 高 | 高 | 中低 |
| 出错成本 | 低(人有最终决定权) | 低(路径固定) | 中高(AI可能走错路) |
| 典型场景 | 写代码、写文章 | 审批流、数据ETL | 调研、自动化执行 |
| 技术门槛 | 低 | 中 | 高 |
选择原则:
- 任务步骤固定、不需要判断 → 工作流
- 任务需要人的创造力和判断力 → Copilot
- 任务复杂、需要AI自主决策和工具调用 → Agent
- 不确定 → 先用工作流或Copilot,别上来就Agent
不要为了Agent而Agent
生产环境中大部分AI应用是”工作流为主,少量Agent”的混合架构。核心业务流程用工作流保证确定性和可审计性,需要灵活决策的环节嵌入Agent。全Agent架构在生产中是反模式。
常见误区:认为Agent一定比工作流”高级”。实际上,一个稳定可靠的工作流,远胜过一个时不时会跑偏的Agent。技术选型要回到业务需求,而不是追求概念上的先进。
练习:判断以下场景分别适合Copilot、工作流还是Agent:
- 每天自动汇总10个团队成员的日报,生成周报
- 写技术方案时AI帮你检查逻辑漏洞
- 让AI自动调研竞品定价策略并生成分析报告
- 客户咨询”我的订单到哪了”时自动查询物流
(参考答案:1-工作流,2-Copilot,3-Agent,4-工作流+简单Agent)
检查点:你能用一句话说清Agent和工作流的本质区别吗?
Agent根据情况自主决策下一步做什么;工作流按预设路径走,不因情况变化而调整。
第2章 Function Calling与MCP
本章目标:让AI从”只会说话”进化到”能调用工具”,掌握Function Calling和MCP两种工具调用方式。
2.1 Function Calling:让AI能”动手”
Function Calling的核心机制:
AI不直接执行任何操作,而是输出结构化的”工具调用请求”,由你的代码来执行。
流程图:
用户: "上海明天天气怎么样?"
↓
AI 思考: 我需要调用天气API获取数据
↓
AI 输出: {"function": "get_weather", "arguments": {"city": "上海", "date": "明天"}}
↓
你的代码: 解析请求,调用真实天气API
↓
返回结果: {"temperature": 22, "condition": "多云", "wind": "东南风3级"}
↓
AI 基于结果回答: "上海明天多云,气温22度,东南风3级,适合出行。"
这个设计有一个关键的安全理念:AI只有”建议权”,没有”执行权”。 你的代码可以做校验、限流、审计,确保AI不会做出危险操作。
2.2 完整代码示例
下面是一个可运行的完整示例,基于OpenAI API:
import openai
import json
# ===== 第一步:定义可用工具 =====
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的天气预报",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称,如'上海'、'北京'"
},
"date": {
"type": "string",
"description": "日期,如'今天'、'明天'、'2025-06-01'"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "查询股票实时价格",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "股票代码,如'0700.HK'(腾讯)"
}
},
"required": ["symbol"]
}
}
}
]
# ===== 第二步:实现工具执行逻辑 =====
def execute_tool(name, args):
"""
实际项目中,这里调用真实的API。
示例中用模拟数据演示。
"""
if name == "get_weather":
# 真实场景:调用天气API(如和风天气、OpenWeatherMap)
weather_db = {
"上海": {"temp": 25, "condition": "多云", "wind": "东南风3级"},
"北京": {"temp": 28, "condition": "晴", "wind": "北风2级"},
}
city = args.get("city", "上海")
data = weather_db.get(city, {"temp": 20, "condition": "未知", "wind": "无"})
return json.dumps({"city": city, **data}, ensure_ascii=False)
elif name == "get_stock_price":
# 真实场景:调用股票API(如东方财富、Yahoo Finance)
stock_db = {
"0700.HK": {"name": "腾讯", "price": 385.2, "change": "+1.5%"},
"AAPL": {"name": "Apple", "price": 195.8, "change": "-0.3%"},
}
symbol = args.get("symbol", "")
data = stock_db.get(symbol, {"name": "未知", "price": 0, "change": "0%"})
return json.dumps({"symbol": symbol, **data}, ensure_ascii=False)
return json.dumps({"error": f"未知工具: {name}"})
# ===== 第三步:Agent主循环 =====
def run_agent(user_message, max_tool_calls=5):
"""完整的Agent循环:对话 → 工具调用 → 结果返回 → 继续对话"""
messages = [
{
"role": "system",
"content": "你是一个智能助手,可以查询天气和股票信息。用中文回答。"
},
{
"role": "user",
"content": user_message
}
]
tool_call_count = 0
while tool_call_count < max_tool_calls:
response = openai.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools
)
msg = response.choices[0].message
messages.append(msg)
# 如果AI决定调用工具
if msg.tool_calls:
for tc in msg.tool_calls:
tool_call_count += 1
args = json.loads(tc.function.arguments)
print(f"🔧 调用工具: {tc.function.name}({args})")
# 执行工具
result = execute_tool(tc.function.name, args)
print(f"📊 结果: {result}")
# 把结果返回给AI
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
else:
# AI直接回复,结束循环
return msg.content
return "⚠️ 达到工具调用上限"
# ===== 运行 =====
if __name__ == "__main__":
answer = run_agent("上海明天天气怎么样?顺便帮我看看腾讯的股票价格。")
print(f"\n💬 AI回答: {answer}")运行输出:
🔧 调用工具: get_weather({"city": "上海", "date": "明天"})
📊 结果: {"city": "上海", "temp": 25, "condition": "多云", "wind": "东南风3级"}
🔧 调用工具: get_stock_price({"symbol": "0700.HK"})
📊 结果: {"symbol": "0700.HK", "name": "腾讯", "price": 385.2, "change": "+1.5%"}
💬 AI回答: 上海明天多云,气温25度,东南风3级,适合出行。腾讯控股(0700.HK)目前报价385.2港元,涨幅1.5%。
2.3 MCP协议:标准化的工具接口
Function Calling有一个实际问题:每个AI平台的工具定义格式不同。
- OpenAI用
tools参数,格式是特定的JSON结构 - Anthropic Claude用
tool_use块,格式不同 - Google Gemini又是一种格式
这导致:你为OpenAI写的工具定义,在Claude上用不了,得重写一遍。
MCP(Model Context Protocol)要解决的问题:定义一套标准协议,让工具只实现一次,所有AI平台都能调用。
类比理解:
- Function Calling = 每个国家用自己的电源插头标准
- MCP = USB-C统一接口,一个充电器到处用
MCP架构:
AI应用(Claude Code / Cursor / 其他)
↕ MCP标准协议
MCP Server(工具服务端)
↕
外部资源(数据库 / API / 文件系统 / 搜索引擎)
MCP Server提供四类能力:
| 能力 | 说明 | 示例 |
|---|---|---|
| Tools | 可调用的函数 | 查数据库、发邮件、执行代码 |
| Resources | 可读取的数据源 | 文件内容、数据库记录 |
| Prompts | 预定义提示词模板 | 特定任务的prompt |
| Sampling | 让Server请求LLM完成子任务 | 复杂推理链中的中间步骤 |
截至2026年,Claude Code、Cursor等主流AI Coding工具已经原生支持MCP。这意味着你写一个MCP Server,就能在多个工具中复用。
2.4 局限性和注意事项
| 局限 | 具体表现 | 应对策略 |
|---|---|---|
| AI可能选错工具 | 复杂场景下误判该调哪个函数 | 工具描述要精确、减少歧义、给示例 |
| 参数不合规 | AI可能生成不存在的城市名、无效日期 | 在execute_tool中做参数校验和兜底 |
| 幻觉工具名 | AI可能编造一个不存在的工具名 | 只返回预定义的工具列表,其他一律拒绝 |
| 调用链过长 | 一个任务可能触发5-10次工具调用 | 设置max_tool_calls上限 |
| 安全风险 | AI可能尝试调用删除数据等危险操作 | 做工具级别权限控制,敏感操作需人工确认 |
| 成本控制 | 每次工具调用都消耗token | 缓存高频调用结果,限制单次对话token上限 |
安全第一原则
Function Calling的本质是”AI提需求,你的代码执行”。永远不要让AI直接执行任何操作。中间必须有一层你的代码来做:参数校验、权限检查、操作审计、限流保护。
练习:给上面的Agent添加一个新工具 send_email(to, subject, body),然后测试”查上海天气并发到 test@example.com”。
检查点:Function Calling中,AI直接执行操作吗?
不是。AI只输出”我想调用什么函数、参数是什么”,由你的代码负责执行。
第3章 ReAct模式与CoT推理
本章目标:掌握让Agent”边想边做”的ReAct模式,以及CoT推理链的实现。
3.1 ReAct = Reasoning + Acting
上一章的Agent是”一步到位”的——用户问什么,AI直接调用工具给出答案。但真实任务往往更复杂:
“帮我调研Notion、Obsidian、Logseq三个笔记软件的优缺点,做个对比表。”
这个任务需要:搜索Notion → 搜索Obsidian → 搜索Logseq → 整理对比 → 生成表格。每一步都依赖上一步的结果。
ReAct模式:让Agent交替进行推理(Thought)和行动(Action),每一步观察结果后再决定下一步做什么。
用户: "调研三个笔记软件的对比"
Thought 1: 用户要对比三个笔记软件,我需要分别搜索它们的信息。
Action 1: search("Notion features pros cons 2025")
Observation 1: Notion是All-in-one工作空间,优点是数据库功能强大...
Thought 2: Notion的信息拿到了,接下来搜索Obsidian。
Action 2: search("Obsidian features pros cons 2025")
Observation 2: Obsidian是本地优先的Markdown笔记,优点是双向链接...
Thought 3: 现在搜索Logseq。
Action 3: search("Logseq features pros cons 2025")
Observation 3: Logseq是大纲式笔记,优点是日志工作流...
Thought 4: 三个软件的信息都收集完了,现在整理对比表。
Final Answer: [生成Notion vs Obsidian vs Logseq对比表格]
ReAct vs 直接回答的核心区别:
| 维度 | 直接回答 | ReAct模式 |
|---|---|---|
| 推理过程 | 隐藏在AI内部 | 显式展示,可审计 |
| 工具使用 | 一次性调用 | 按需逐步调用 |
| 纠错能力 | 无(错了就错了) | 有(观察到错误可以调整) |
| 可控性 | 低 | 高(每步都能观察) |
| Token消耗 | 少 | 多(多了推理步骤) |
3.2 完整ReAct Agent代码实现
import openai
import json
REACT_SYSTEM_PROMPT = """你是一个使用ReAct模式工作的研究助理。
对于每个任务,严格按照以下格式思考和行动:
Question: 用户的原始问题
Thought: 你当前的思考(要做什么、为什么)
Action: 要调用的工具名称(从可用工具列表中选择)
Action Input: 工具参数(JSON格式字符串)
Observation: [系统自动填入工具返回结果]
你可以进行多轮 Thought/Action/Action Input,直到你有足够信息。
当你准备好回答时,使用:
Thought: 我现在有足够的信息来回答问题了
Final Answer: [你的完整回答]
可用工具:
- search(query): 搜索网络信息,参数query是搜索关键词
- calculate(expression): 计算数学表达式
- read_file(path): 读取文件内容"""
# 工具执行器
def execute_react_tool(name, args_str):
"""执行ReAct Agent调用的工具"""
try:
args = json.loads(args_str)
except json.JSONDecodeError:
return f"参数解析错误: {args_str}"
if name == "search":
# 实际项目中调用搜索API
return f"[搜索结果] 关于'{args.get('query', '')}'的信息:..."
elif name == "calculate":
try:
result = eval(args.get("expression", "0"))
return str(result)
except Exception as e:
return f"计算错误: {e}"
elif name == "read_file":
# 实际项目中读真实文件
return f"[文件内容] {args.get('path', '')}的内容..."
return f"未知工具: {name}"
def run_react_agent(question, max_iterations=8):
"""ReAct Agent主循环"""
messages = [
{"role": "system", "content": REACT_SYSTEM_PROMPT},
{"role": "user", "content": question}
]
for i in range(max_iterations):
response = openai.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0 # 低温度保证推理稳定
)
output = response.choices[0].message.content
messages.append({"role": "assistant", "content": output})
# 检查是否已完成
if "Final Answer:" in output:
final = output.split("Final Answer:")[-1].strip()
print(f"✅ 完成({i+1}轮迭代)")
return final
# 解析并执行Action
if "Action:" in output and "Action Input:" in output:
try:
action_line = output.split("Action:")[-1].split("\n")[0].strip()
input_line = output.split("Action Input:")[-1].split("\n")[0].strip()
print(f" 轮次{i+1}: {action_line}")
observation = execute_react_tool(action_line, input_line)
messages.append({
"role": "user",
"content": f"Observation: {observation}"
})
except Exception as e:
messages.append({
"role": "user",
"content": f"Observation: 执行出错 - {e}"
})
else:
# AI没有调用工具也没有Final Answer,可能是纯推理
continue
return "⚠️ 达到最大迭代次数,未能完成"
# 运行示例
if __name__ == "__main__":
result = run_react_agent(
"比较React、Vue、Svelte三个前端框架在2025年的生态和性能"
)
print(result)3.3 CoT(Chain of Thought)推理链
CoT是ReAct的基础——让AI把推理过程显式写出来,而不是直接跳到结论。
为什么CoT有效?因为大模型的推理能力来自”逐token生成”。让AI写出推理步骤,相当于给它更多的”思考时间”和”工作记忆”。
三种CoT实现方式:
| 方式 | 做法 | 效果 | 适用 |
|---|---|---|---|
| Zero-shot CoT | 提示词末尾加”请一步步思考” | 提升30-50% | 简单推理 |
| Few-shot CoT | 给几个带推理过程的示例 | 提升50-80% | 特定领域 |
| Auto-CoT | 让AI自己生成推理示例 | 接近Few-shot | 通用推理 |
Zero-shot CoT示例:
用户: 一个商店打8折后是320元,原价多少?
普通AI: 原价是400元。(可能算错)
加了CoT: 原价 × 0.8 = 320,所以原价 = 320 ÷ 0.8 = 400元。
3.4 推理模式演进路线
从简单到复杂的推理模式演进:
CoT (Chain of Thought)
→ 纯推理链,不调用工具
→ 适合:数学题、逻辑推理
ReAct (Reasoning + Acting)
→ 推理 + 工具调用交替
→ 适合:需要搜索/计算的复杂任务
Reflexion
→ ReAct + 自我反思
→ Agent执行后回顾结果,发现错误自动重试
→ 适合:需要自我纠错的任务
LATS (Language Agent Tree Search)
→ ReAct + 多路径搜索
→ 同时探索多条路径,选最优
→ 适合:决策空间大的任务(如代码生成)
什么时候用哪种模式
- 只需要AI”想清楚”(数学、逻辑)→ CoT
- 需要AI”边想边做”(调研、操作工具)→ ReAct
- 任务容易出错需要自我纠正 → Reflexion
- 不确定最佳路径需要探索 → LATS
常见坑点:ReAct模式消耗的token远多于直接回答,因为每一步都要输出推理过程。生产环境中要监控token用量,设置合理的迭代上限。
练习:用上面的ReAct Agent处理一个真实任务:“帮我分析Python和Go在2025年的就业市场对比”。观察它的推理步骤。
检查点:ReAct模式比直接调用工具好在哪里?
ReAct让Agent能根据中间结果调整策略,而不是一次性猜测答案。相当于”摸着石头过河”vs”蒙着眼跳”。
第4章 记忆系统设计
本章目标:理解Agent为什么需要记忆,掌握短期记忆和长期记忆的设计与实现。
4.1 为什么Agent需要记忆
没有记忆的Agent像一条金鱼——每次对话从零开始。真实场景中的记忆需求:
| 场景 | 需要的记忆类型 | 示例 |
|---|---|---|
| 多轮对话 | 短期记忆 | ”我刚才说了什么?“ |
| 用户偏好 | 长期记忆 | ”记住我喜欢用Python” |
| 任务进度 | 工作记忆 | ”调研做到哪一步了?“ |
| 知识积累 | 长期知识 | ”上次项目的经验教训” |
4.2 短期记忆实现
短期记忆 = 当前对话的上下文,受模型token上限约束。
三种管理策略:
策略一:完整历史(最简单)
# 直接把所有历史消息传给模型
# 优点:信息完整 缺点:超出token上限就崩
messages = [
{"role": "user", "content": "我在上海"},
{"role": "assistant", "content": "好的,记住你在上海"},
# ... 随着对话越来越长 ...
{"role": "user", "content": "明天出门带伞吗?"},
]策略二:滑动窗口(最常用)
def sliding_window(messages, keep_rounds=10):
"""保留最近N轮对话,丢弃更早的"""
# 每轮 = 1条user + 1条assistant
return messages[-(keep_rounds * 2):]策略三:摘要压缩(最优雅)
def compress_history(messages, llm_client, keep_recent=6):
"""把旧对话压缩成摘要,保留最近几轮完整"""
if len(messages) <= keep_recent:
return messages
old_messages = messages[:-keep_recent]
recent_messages = messages[-keep_recent:]
# 用LLM生成摘要
summary_prompt = f"请用3句话总结以下对话的关键信息:\n{old_messages}"
summary = llm_client.chat(summary_prompt)
return [
{"role": "system", "content": f"[对话历史摘要] {summary}"}
] + recent_messages4.3 长期记忆实现
长期记忆 = 跨对话持久化的信息,保存在外部存储中。
import json
import os
from datetime import datetime
class LongTermMemory:
"""Agent长期记忆系统"""
def __init__(self, storage_path="agent_memory.json"):
self.path = storage_path
self.data = self._load()
def _load(self):
if os.path.exists(self.path):
with open(self.path, "r", encoding="utf-8") as f:
return json.load(f)
return {
"user_profile": {}, # 用户画像:名字、偏好、习惯
"events": [], # 事件记忆:重要对话、决策、结果
"knowledge": [], # 知识积累:学到的规律、经验
"feedback": [] # 反馈记录:用户对回答的评价
}
def save(self):
with open(self.path, "w", encoding="utf-8") as f:
json.dump(self.data, f, ensure_ascii=False, indent=2)
def update_profile(self, key, value):
"""更新用户画像"""
self.data["user_profile"][key] = value
self.save()
def add_event(self, event, tags=None):
"""记录重要事件"""
self.data["events"].append({
"time": datetime.now().isoformat(),
"event": event,
"tags": tags or []
})
# 只保留最近100条事件
if len(self.data["events"]) > 100:
self.data["events"] = self.data["events"][-100:]
self.save()
def add_knowledge(self, fact, source="conversation"):
"""积累知识"""
self.data["knowledge"].append({
"fact": fact,
"source": source,
"time": datetime.now().isoformat()
})
self.save()
def get_profile_summary(self):
"""生成用户画像摘要文本"""
profile = self.data["user_profile"]
if not profile:
return "暂无用户信息。"
lines = ["已知用户信息:"]
for k, v in profile.items():
lines.append(f"- {k}: {v}")
return "\n".join(lines)
def get_recent_events(self, n=5):
"""获取最近N条事件"""
return self.data["events"][-n:]
def search_events(self, keyword, limit=5):
"""简单关键词搜索(生产环境应改为向量搜索)"""
results = []
for event in reversed(self.data["events"]):
if keyword in event["event"]:
results.append(event)
if len(results) >= limit:
break
return results
def build_context(self, current_query):
"""为当前对话构建记忆上下文"""
context_parts = []
# 用户画像
if self.data["user_profile"]:
context_parts.append(self.get_profile_summary())
# 相关事件
relevant = self.search_events(current_query, limit=3)
if relevant:
context_parts.append("相关历史:")
for e in relevant:
context_parts.append(f"- [{e['time'][:10]}] {e['event']}")
return "\n".join(context_parts) if context_parts else ""4.4 意图识别:决定什么该记什么该忘
不是所有对话都值得记忆。需要一个”门卫”来判断。
MEMORY_GATE_PROMPT = """分析用户最新消息,判断是否需要更新记忆。
当前用户画像:{profile}
用户消息:{message}
请返回JSON(不要其他内容):
- 包含新个人信息(姓名、位置、偏好、习惯)→ {"action": "update_profile", "key": "...", "value": "..."}
- 包含重要事件或决策 → {"action": "add_event", "event": "..."}
- 包含有价值的知识或规律 → {"action": "add_knowledge", "fact": "..."}
- 临时闲聊或不重要 → {"action": "skip"}"""
def process_memory_gate(message, memory, llm_client):
"""每轮对话后判断是否更新记忆"""
import json as _json
prompt = MEMORY_GATE_PROMPT.format(
profile=_json.dumps(memory.data["user_profile"], ensure_ascii=False),
message=message
)
result = llm_client.chat(prompt)
try:
parsed = _json.loads(result.strip())
action = parsed.get("action", "skip")
if action == "update_profile":
memory.update_profile(parsed["key"], parsed["value"])
print(f"📝 记住: {parsed['key']} = {parsed['value']}")
elif action == "add_event":
memory.add_event(parsed["event"])
print(f"📅 记录事件: {parsed['event'][:30]}...")
elif action == "add_knowledge":
memory.add_knowledge(parsed["fact"])
print(f"💡 积累知识: {parsed['fact'][:30]}...")
# skip = 不记忆
except _json.JSONDecodeError:
pass # 解析失败不影响主流程把记忆集成到Agent主循环中:
def run_agent_with_memory(user_message, memory, messages_history):
# 1. 构建记忆上下文
memory_ctx = memory.build_context(user_message)
# 2. 组装完整提示
system_prompt = f"你是一个智能助手。\n{memory_ctx}" if memory_ctx else "你是一个智能助手。"
# 3. 管理短期记忆
if len(messages_history) > 30:
messages_history = compress_history(messages_history, llm)
messages = [{"role": "system", "content": system_prompt}] + messages_history
messages.append({"role": "user", "content": user_message})
# 4. 调用LLM(加上工具调用逻辑)
response = openai.chat.completions.create(model="gpt-4o", messages=messages)
reply = response.choices[0].message.content
# 5. 更新长期记忆
process_memory_gate(user_message, memory, llm)
return reply记忆的成本权衡
记忆越多,每次调用token越多,成本越高。生产环境要做记忆分级:核心信息(用户画像)永久保留,事件记忆定期清理(保留最近100条),临时信息不存储。同时要做记忆去重——同一件事不要记两遍。
练习:给Agent添加”记住我的咖啡偏好”功能。当用户说”我要一杯美式”时自动记住偏好,下次说”帮我点咖啡”时自动推荐美式。
检查点:短期记忆和长期记忆的核心区别?
短期记忆在对话内,受token限制,对话结束就消失。长期记忆跨对话,持久化存储,但需要额外的检索和管理机制。
第5章 工作流工具选型
本章目标:掌握三大工作流工具的特点和适用场景,能做出正确的选型决策。
5.1 Coze:字节跳动的AI工作流平台
Coze是字节跳动推出的AI Bot构建平台,面向非技术用户。
| 特点 | 说明 |
|---|---|
| 上手难度 | 极低,可视化拖拽,无需编程 |
| 工作流能力 | 支持条件分支、循环、变量传递 |
| 知识库 | 内置RAG,支持文档/网页/数据库 |
| 插件市场 | 丰富,有搜索、天气、数据库等插件 |
| 部署方式 | 仅云端托管,不支持私有部署 |
| 价格 | 有免费额度,超出按量付费 |
| 集成能力 | 支持API调用外部服务 |
适合场景:快速验证想法、搭建客服机器人、内容生成工具。团队没有开发资源时首选。
不适合:需要私有部署、处理敏感数据、需要深度定制。
5.2 Dify / FastGPT:开源AI应用平台
Dify是一个开源的LLM应用开发平台,面向有技术背景的团队。
| 特点 | 说明 |
|---|---|
| 上手难度 | 中等,需要理解Agent/工作流概念 |
| 工作流能力 | 强大,支持复杂编排和子流程 |
| 知识库 | 完整RAG支持,可自定义chunk策略 |
| Agent能力 | 支持Function Calling、ReAct模式 |
| 部署方式 | Docker私有部署,或使用云服务 |
| 开源协议 | Apache 2.0 |
| API | 完善的REST API,支持Webhook |
FastGPT更专注于知识库场景,在RAG方面做了深度优化(更好的chunk策略、重排序)。
适合场景:企业级应用、需要私有部署、定制化需求强、有开发团队。
5.3 n8n:通用自动化工作流
n8n是一个通用自动化平台,AI只是它的能力之一。
| 特点 | 说明 |
|---|---|
| 上手难度 | 中等,需要理解工作流概念 |
| 工作流能力 | 非常强大,支持复杂条件/循环/子流程 |
| 集成数量 | 400+内置集成(CRM、邮件、数据库等) |
| AI节点 | 支持调用LLM、做AI工作流 |
| 部署方式 | 云端或私有部署(Docker) |
| 开源协议 | 公平使用许可(Fair Code) |
适合场景:需要连接多个非AI系统(如CRM+邮件+数据库+AI)的自动化流程。
5.4 选型决策矩阵
决策树:
你的需求是什么?
│
├─ 快速上线,团队无开发资源
│ └─ → Coze
│
├─ 需要私有部署 / 深度定制
│ ├─ 以知识库/RAG为核心
│ │ └─ → FastGPT 或 Dify
│ └─ 需要完整Agent编排
│ └─ → Dify
│
├─ 需要连接多个外部系统做自动化
│ └─ → n8n(可搭配Dify做AI后端)
│
└─ 需要完全自定义的Agent行为
└─ → 自建(LangChain / LlamaIndex + Python)
五维对比表:
| 维度 | Coze | Dify | FastGPT | n8n | 自建 |
|---|---|---|---|---|---|
| 上手速度 | ★★★★★ | ★★★ | ★★★ | ★★★ | ★ |
| 灵活性 | ★★ | ★★★★ | ★★★ | ★★★★ | ★★★★★ |
| Agent能力 | ★★★ | ★★★★ | ★★ | ★★★ | ★★★★★ |
| 知识库质量 | ★★★ | ★★★★ | ★★★★★ | ★★ | 自定义 |
| 私有部署 | ✗ | ✓ | ✓ | ✓ | ✓ |
| 运维成本 | 低 | 中 | 中 | 中 | 高 |
选型的真正原则
不要选”最好的工具”,要选”团队能驾驭的工具”。一个团队能用好的Coze应用,比一个没人维护的Dify应用有价值得多。先跑通MVP,再考虑迁移到更灵活的平台。
练习:用上面的决策树分析你的工作场景,写下应该选哪个工具及理由。
检查点:什么场景必须自建而不是用平台?
需要完全自定义的Agent行为(平台不支持的工具调用模式)、对数据安全有极高要求(不能经过第三方)、或者现有平台无法满足的特殊业务逻辑。
第6章 Mini-Manus实战
本章目标:理解Manus架构,实现一个能自主完成任务的Mini-Manus。
6.1 Manus架构解析
Manus是一个能自主完成复杂任务的AI Agent系统,核心架构:
┌─────────────────────────────────────────┐
│ Manus 核心架构 │
│ │
│ ┌──────────┐ ┌──────────┐ │
│ │ 规划器 │───→│ 执行器 │ │
│ │ Planner │ │ Executor │ │
│ └──────────┘ └──────────┘ │
│ │ │ │
│ ┌────┴────┐ ┌─────┴────┐ │
│ │ 任务拆解 │ │ 工具调度 │ │
│ │ 依赖分析 │ │ 结果收集 │ │
│ └─────────┘ └──────────┘ │
│ │ │ │
│ ┌────┴────────────┐ │ │
│ │ 四大核心能力 │ │ │
│ │ ┌──────┐┌──────┐│ │ │
│ │ │浏览器││代码 ││ │ │
│ │ │Browser│Sandbox││ │ │
│ │ └──────┘└──────┘│ │ │
│ │ ┌──────┐┌──────┐│ │ │
│ │ │文件 ││记忆 ││ │ │
│ │ │Files ││Memory ││ │ │
│ │ └──────┘└──────┘│ │ │
│ └─────────────────┘ │ │
└─────────────────────────────────────────┘
四大核心能力:
| 能力 | 作用 | 实现方式 |
|---|---|---|
| 浏览器操作 | 打开网页、点击、输入、截图 | Playwright/Puppeteer |
| 代码执行 | 写代码并运行 | Docker沙箱 |
| 文件管理 | 创建、读取、编辑文件 | 文件系统API |
| 任务规划 | 拆解复杂任务为步骤 | LLM + 规划提示词 |
核心工作循环:
接收任务 → 拆解为步骤 → 逐步执行(调用工具)
→ 观察结果 → 判断是否完成 → 未完成则调整继续
→ 完成则汇总结果
6.2 Mini-Manus完整实现
下面实现一个简化版Manus:去掉代码沙箱,保留浏览器、文件、搜索三个能力。
import openai
import json
import os
client = openai.OpenAI()
class MiniManus:
"""简化版Manus Agent"""
def __init__(self, work_dir="./manus_workspace"):
self.work_dir = work_dir
os.makedirs(work_dir, exist_ok=True)
self.tools = self._define_tools()
self.conversation = []
def _define_tools(self):
return [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索网络获取信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "browse_url",
"description": "访问指定URL并获取页面内容",
"parameters": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "要访问的URL"}
},
"required": ["url"]
}
}
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "创建或写入文件",
"parameters": {
"type": "object",
"properties": {
"filename": {"type": "string", "description": "文件名"},
"content": {"type": "string", "description": "文件内容"}
},
"required": ["filename", "content"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取文件内容",
"parameters": {
"type": "object",
"properties": {
"filename": {"type": "string", "description": "文件名"}
},
"required": ["filename"]
}
}
}
]
def _execute_tool(self, name, args):
"""执行工具调用"""
if name == "search_web":
# 实际实现:调用搜索API(Tavily / SerpAPI / Bing Search)
return json.dumps({
"results": [
f"[模拟] {args['query']}的搜索结果1:...",
f"[模拟] {args['query']}的搜索结果2:..."
]
}, ensure_ascii=False)
elif name == "browse_url":
# 实际实现:用Playwright抓取页面
return json.dumps({
"url": args["url"],
"content": f"[模拟] {args['url']}的页面内容..."
}, ensure_ascii=False)
elif name == "write_file":
filepath = os.path.join(self.work_dir, args["filename"])
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, "w", encoding="utf-8") as f:
f.write(args["content"])
return json.dumps({"status": "ok", "path": filepath})
elif name == "read_file":
filepath = os.path.join(self.work_dir, args["filename"])
if os.path.exists(filepath):
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
return json.dumps({"status": "ok", "content": content})
return json.dumps({"error": "文件不存在"})
return json.dumps({"error": f"未知工具: {name}"})
def run(self, task, max_steps=15):
"""执行任务"""
print(f"🎯 任务: {task}\n")
self.conversation = [
{
"role": "system",
"content": """你是一个能自主完成任务的AI助手。
你可以:
- search_web: 搜索网络信息
- browse_url: 访问网页获取详细内容
- write_file: 创建文件保存结果
- read_file: 读取已创建的文件
工作方式:
1. 先规划任务步骤
2. 逐步执行,每步调用一个工具
3. 根据结果决定下一步
4. 最后将结果写入文件
每次只调用一个工具,等结果后再决定下一步。"""
},
{"role": "user", "content": task}
]
for step in range(max_steps):
response = client.chat.completions.create(
model="gpt-4o",
messages=self.conversation,
tools=self.tools
)
msg = response.choices[0].message
self.conversation.append(msg)
if msg.tool_calls:
for tc in msg.tool_calls:
args = json.loads(tc.function.arguments)
result = self._execute_tool(tc.function.name, args)
self.conversation.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
print(f" 步骤{step+1}: {tc.function.name}({json.dumps(args, ensure_ascii=False)[:80]})")
else:
print(f"\n✅ 任务完成!")
return msg.content
return "⚠️ 达到最大步骤数"
# ===== 运行 =====
if __name__ == "__main__":
agent = MiniManus()
result = agent.run(
"调研2025年最流行的三个前端框架(React/Vue/Svelte),"
"分别搜索它们的最新特性,然后生成一个对比报告保存到 frontend-comparison.md"
)
print(result)6.3 浏览器操作实现(Playwright)
Mini-Manus的浏览器能力需要Playwright实现:
# 安装:pip install playwright && playwright install chromium
from playwright.sync_api import sync_playwright
def browse_page(url, action="read", selector=None, value=None):
"""用Playwright操作浏览器"""
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(url, timeout=30000)
if action == "read":
# 提取页面文本内容(去掉HTML标签)
content = page.inner_text("body")
browser.close()
return content[:5000] # 限制长度,避免token爆炸
elif action == "click":
page.click(selector)
page.wait_for_load_state("networkidle")
content = page.inner_text("body")
browser.close()
return content[:5000]
elif action == "input":
page.fill(selector, value)
page.press(selector, "Enter")
page.wait_for_load_state("networkidle")
content = page.inner_text("body")
browser.close()
return content[:5000]
elif action == "screenshot":
screenshot_bytes = page.screenshot(full_page=True)
browser.close()
return screenshot_bytes # 返回截图bytes6.4 局限和改进方向
| 局限 | 原因 | 改进方向 |
|---|---|---|
| 无法执行代码 | 没有沙箱环境 | 加Docker沙箱执行Python/JS |
| 浏览器操作不智能 | 只能做基础read/click/input | 加截图+视觉理解(多模态) |
| 没有长期记忆 | 每次任务独立运行 | 集成第4章的LongTermMemory |
| 错误恢复能力弱 | 只能重试 | 加Reflexion自我反思循环 |
| 并行能力差 | 串行执行工具 | 支持并行工具调用 |
| 搜索质量低 | 模拟数据 | 接入真实搜索API(Tavily) |
Mini-Manus的教育意义
你不需要做出一个和Manus一样强大的系统。Mini-Manus的价值是让你理解Agent的核心机制:规划 → 执行 → 观察 → 调整。理解了这个循环,你就能在此基础上逐步增强——加沙箱、加记忆、加多Agent协作。
练习:给Mini-Manus添加一个 send_notification(message) 工具,任务完成时通过webhook发送通知。
检查点:Mini-Manus的核心工作循环是什么?
接收任务 → 规划步骤 → 调用工具执行 → 观察结果 → 判断完成或调整 → 循环。
第7章 多Agent协作系统
本章目标:理解多Agent协作架构,实现一个基础的多Agent系统。
7.1 多Agent协作的设计哲学
核心思想:让多个专精Agent分工协作,而不是用一个通用Agent硬扛所有事。
类比:一个公司不是一个人干所有事,而是有产品经理、设计师、开发、测试各司其职。每个人做自己擅长的事,通过协作完成复杂项目。
多Agent协作的三种模式:
模式一:串行Pipeline(A做完→B做→C做)
用户 → 研究员 → 写作者 → 审查员 → 最终输出
模式二:并行协作(A和B同时做→合并)
用户 → [研究员A + 研究员B] → 合并器 → 输出
模式三:讨论/对抗(A和B对同一问题讨论)
用户 → 支持方 ↔ 反对方 → 裁判 → 输出
7.2 Agent基类和编排器
import openai
import json
client = openai.OpenAI()
class Agent:
"""单个Agent"""
def __init__(self, name, role_description, model="gpt-4o"):
self.name = name
self.role = role_description
self.model = model
def run(self, task, context=""):
"""执行任务"""
messages = [
{"role": "system", "content": self.role},
{"role": "user", "content": f"{'背景:' + context + chr(10) if context else ''}任务:{task}"}
]
response = client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7
)
return response.choices[0].message.content
class MultiAgentOrchestrator:
"""多Agent编排器"""
def __init__(self):
self.agents = {
"researcher": Agent(
"研究员",
"你是一个专业的信息研究员。擅长搜索、整理、分析信息。输出结构化的调研结果。"
),
"writer": Agent(
"写作者",
"你是一个专业的内容写作者。擅长将调研结果转化为结构清晰、可读性强的文章。"
),
"reviewer": Agent(
"审查员",
"你是一个严格的质量审查员。检查文章的准确性、完整性、逻辑性。给出具体的修改建议和1-10分评分。"
)
}
def run_pipeline(self, topic):
"""串行Pipeline模式:调研 → 写作 → 审查 → 修改"""
print("=" * 50)
print(f"📋 开始处理任务: {topic}")
print("=" * 50)
# Step 1: 调研
print("\n🔍 Step 1: 研究员调研中...")
research = self.agents["researcher"].run(
f"请对以下主题进行全面调研,列出关键要点、数据、不同观点:{topic}"
)
print(f" 调研完成,收集了{len(research)}字信息")
# Step 2: 写作
print("\n✍️ Step 2: 写作者撰写中...")
draft = self.agents["writer"].run(
"基于调研结果,写一篇结构清晰的文章。包含:标题、摘要、正文(分小节)、结论。",
context=research
)
print(f" 初稿完成,{len(draft)}字")
# Step 3: 审查
print("\n🔎 Step 3: 审查员审核中...")
review = self.agents["reviewer"].run(
"请严格审查以下文章。评价:1)准确性 2)完整性 3)逻辑性 4)可读性。给出评分和具体修改建议。",
context=draft
)
print(f" 审查完成")
print(f" 审查意见:\n{review[:200]}...")
# Step 4: 根据审查修改
print("\n🔄 Step 4: 写作者根据审查修改...")
final = self.agents["writer"].run(
f"根据以下审查意见修改文章。保留优点,改进不足的地方。\n\n审查意见:\n{review}",
context=draft
)
print("\n" + "=" * 50)
print("✅ 全部完成!")
print("=" * 50)
return {
"research": research,
"draft": draft,
"review": review,
"final": final
}
# ===== 运行 =====
if __name__ == "__main__":
system = MultiAgentOrchestrator()
result = system.run_pipeline(
"2025年AI Agent技术的发展趋势和商业落地"
)
# 打印最终结果
print("\n" + "=" * 50)
print("📄 最终文章:")
print("=" * 50)
print(result["final"][:2000]) # 打印前2000字7.3 从Mini版到生产版的差距
| 维度 | Mini版(本章实现) | 生产版(真实系统) |
|---|---|---|
| Agent通信 | 字符串全文传递 | 结构化JSON Schema |
| 错误处理 | 基本try/catch | 重试+降级+熔断+死信队列 |
| 并行执行 | 无 | asyncio多Agent并行 |
| 状态管理 | 内存变量 | 持久化存储+状态机 |
| 上下文管理 | 全量传递(token浪费严重) | 压缩+摘要+检索 |
| 人工介入 | 无 | 随时可中断+审批节点 |
| 可观测性 | 日志+监控+链路追踪 | |
| 安全性 | 无限制 | 工具权限+输出过滤+敏感信息脱敏 |
| Agent数量 | 3个固定 | 动态创建+销毁 |
| 任务队列 | 无 | 优先级队列+超时控制 |
真实的多Agent系统
Claude Code的Agent团队、CrewAI等生产级系统,不仅有多个Agent,还有任务分配器、上下文管理器、质量检查器等大量基础设施。本章学到的是核心骨架,真实系统需要在此基础上加工程化能力:日志、监控、重试、安全。
练习:给MultiAgentOrchestrator添加一个”Editor”Agent,在writer和reviewer之间做一次内容优化(调整结构、控制字数、增强可读性)。
检查点:多Agent协作比单Agent好在哪里,代价是什么?
好处:每个Agent专注做自己擅长的事,角色更清晰,输出更可控。代价:系统复杂度更高、token消耗更多、调试更困难。
第8章 综合实战:搭建一个完整的AI应用
本章目标:从0到1完成一个AI应用,覆盖需求分析→选型→搭建→测试→上线。
8.1 需求分析模板
案例场景:搭建一个”智能客服知识库”——AI基于公司文档回答客户问题。
| 维度 | 具体要求 |
|---|---|
| 目标用户 | 公司客户(非技术人员) |
| 核心场景 | 查产品功能、价格、售后政策 |
| 数据来源 | 产品文档、FAQ、历史工单 |
| 回答质量 | 必须准确,不能编造(幻觉率<5%) |
| 响应速度 | 3秒内返回首字 |
| 并发要求 | 同时100+用户 |
| 部署要求 | 私有部署(数据不能外泄) |
| 降级方案 | AI不可用时回退到关键词匹配 |
架构设计:
用户 → API网关 → Dify(Agent+RAG引擎)
↓
LLM API(GPT-4o / Claude)
↓
工具(查订单状态、查物流)
↓
向量数据库(Weaviate/Milvus)
↓
文档管理系统(原始知识库)
8.2 工具链选择
基于第5章的选型分析,选 Dify + Weaviate + GPT-4o:
| 组件 | 选型 | 选型理由 |
|---|---|---|
| 工作流平台 | Dify | 需要私有部署,有完整Agent能力 |
| LLM | GPT-4o | 中文能力强,延迟可接受 |
| Embedding | text-embedding-3-large | 效果好,支持中文 |
| 向量数据库 | Weaviate | 开源,支持混合搜索(向量+关键词) |
| API网关 | Nginx | 限流+鉴权+负载均衡 |
| 监控 | Langfuse | LLM调用追踪和评估 |
集成步骤:
- 部署Dify(Docker Compose一键部署)
- 在Dify中创建知识库,导入产品文档
- 配置Agent:设置系统提示词+接入工具
- 通过Dify API接入前端页面
- 配置Langfuse监控
8.3 测试方案
# 客服Agent自动化测试
test_cases = [
{
"id": "TC001",
"question": "你们的退货政策是什么?",
"expected_keywords": ["7天", "无理由", "退款"],
"must_not_contain": ["我不确定", "建议您联系", "抱歉我无法"],
"category": "售后"
},
{
"id": "TC002",
"question": "Pro版本多少钱?",
"expected_keywords": ["299", "月", "功能"],
"must_not_contain": [],
"category": "价格"
},
{
"id": "TC003",
"question": "怎么联系你们的客服?",
"expected_keywords": ["电话", "邮箱", "在线"],
"must_not_contain": [],
"category": "联系方式"
}
]
def evaluate_agent(test_cases, agent_endpoint):
"""运行测试并评估结果"""
results = []
for case in test_cases:
# 调用Agent API
answer = call_agent(agent_endpoint, case["question"])
# 评估
keyword_hit = all(kw in answer for kw in case["expected_keywords"])
forbidden_hit = any(kw in answer for kw in case["must_not_contain"])
passed = keyword_hit and not forbidden_hit
results.append({
"id": case["id"],
"category": case["category"],
"question": case["question"],
"answer_preview": answer[:100],
"keyword_hit": keyword_hit,
"no_forbidden": not forbidden_hit,
"passed": passed
})
status = "✅ PASS" if passed else "❌ FAIL"
print(f"{status} [{case['category']}] {case['question'][:30]}...")
# 统计
total = len(results)
passed = sum(r["passed"] for r in results)
print(f"\n通过率: {passed}/{total} ({passed/total*100:.0f}%)")
return results
def call_agent(endpoint, question):
"""调用Agent API(示例)"""
import requests
resp = requests.post(endpoint, json={"query": question})
return resp.json()["answer"]迭代优化清单(按优先级排列):
| 问题 | 诊断 | 优化方案 |
|---|---|---|
| 准确率不够 | 知识库数据缺失 | 补充缺失文档,增加FAQ条目 |
| AI编造信息 | 提示词不够严格 | 加”只基于知识库回答,不知道就说不知道” |
| 回答太长 | 没有字数控制 | 加输出长度限制和结构化要求 |
| 速度太慢 | 检索结果太多 | 减少top-k(5→3),用更快的模型 |
| 用户不满意 | 特定场景回答差 | 分析日志,针对性补充知识 |
8.4 上线检查清单和运维指标
上线前检查清单:
- 知识库数据全部导入并验证(跑过测试用例)
- API限流配置(每用户每分钟10次请求)
- 监控告警(延迟>5s、错误率>5%触发告警)
- 降级方案就绪(LLM不可用时回退关键词匹配)
- 日志记录(每次对话记录question/answer/latency/source)
- 人工接管功能(“转人工”按钮)
- 安全审计(确认不泄露敏感信息)
- 压测通过(100并发,P99延迟<10s)
上线后运维指标:
| 指标 | 目标值 | 监控方式 |
|---|---|---|
| 响应延迟 | P95 < 3秒 | API监控 |
| 回答准确率 | > 90% | 人工抽检 + 自动评估 |
| 幻觉率 | < 5% | 关键词检测 + 人工抽检 |
| 用户满意度 | > 4.0/5.0 | 对话后评分 |
| 转人工率 | < 15% | 按钮点击统计 |
| 知识库命中率 | > 80% | 检索日志分析 |
上线不是终点
AI应用上线后才是真正的开始。建议每周做一次bad case分析:随机抽取20条回答不满意的对话,人工标注问题原因(知识库缺失/检索不准/模型幻觉/提示词不好),按优先级迭代优化。这是一个持续的过程。
练习:为你自己的AI应用场景写一份完整的需求文档(用上面的模板),画出架构图,列出工具选型理由,写5条测试用例。
检查点:AI应用上线后最重要的运维动作是什么?
持续分析bad case,因为AI的行为会随用户输入变化而变化,不能”上线就不管了”。每周抽检+迭代优化是基本节奏。