从零构建企业级AI Agent：5个月踩坑实录，满意度从60%到85%
·
从零构建企业级AI Agent：5个月踩坑实录，满意度从60%到85%
作者按： 去年Q3做智能客服，最开始满意度只有60%，用户骂声一片。调整后达到85%，但这不是"调prompt"的功劳。本文是真实复盘，包含完整代码和踩坑记录。阅读约20分钟，建议边看边试。
引言：我们为什么放弃"直接调GPT-4"？
去年Q3，我带团队做智能客服。最开始的想法特别简单：把用户问题扔给GPT-4，让它回答不就完了？
效果一般。用户满意度只有60%左右。
问题出在哪？我们复盘了100多个失败案例，发现核心问题是：GPT-4只是个"问答机器"，不是"助手"。
具体说：
- 它查不了订单系统（不知道用户买了啥）
- 它执行不了退款操作（没有工具调用能力）
- 它不会反思（给了错误答案也不自知）
后来我们引入了Agent架构——让AI能够自己做计划、调工具、发现错误后纠正——满意度直接飙到85%。
这不是"调prompt"能解决的。这是架构级别的升级。
本文将分享：我们怎么从0到1搭建这套系统、踩了哪些坑、哪些决策现在看都是错的。
技术栈： Python 3.11 + LangChain 0.1.x + GPT-4 Turbo + FAISS
一、AI Agent到底是什么？
1.1 别被名词忽悠了
很多人（包括半年前的我）以为："调用一次GPT API"就是Agent了。
不对。
真正的Agent，和普通LLM调用的区别：
普通调用：
你：帮我分析竞品的定价策略
GPT：好的，根据市场情况，我建议...
（然后瞎编一通，因为它根本没查数据）
Agent：
你：帮我分析竞品的定价策略
Agent：
1. 先去爬竞品网站的价格数据
2. 发现数据不全，再查一下用户评价
3. 用Python做价格分布分析
4. 生成可视化图表
5. 写报告，附上数据来源
看出区别没？Agent会"自己想办法"完成任务，而不只是"回答问题"。
我们团队总结的三大核心能力：
| 能力 | 人话翻译 | 实际例子 |
|---|---|---|
| 自主规划 | AI自己拆解任务 | 用户问"分析竞品定价"，AI自动规划：搜索→爬取→分析→写报告 |
| 工具调用 | AI能"动手"干活 | 查天气API、跑SQL、调内部系统、爬网页 |
| 反思纠错 | AI发现错了会改 | 数据不全→去补数据；代码报错→去修代码 |
1.2 主流框架对比（我们踩过的坑）
项目启动时，我们在框架选择上纠结了很久。最后决定：快速原型用LangChain，生产环境自研。
为什么？看对比：
| 框架 | 适合干嘛 | 优点 | 坑点 | 我们的评价 |
|---|---|---|---|---|
| LangChain | 快速验证想法 | 生态好，社区活跃 | 抽象层次太高，调试想骂人 | ⭐⭐⭐⭐ 适合原型 |
| LlamaIndex | 知识库问答 | 数据处理牛逼，RAG效果好 | Agent能力弱，工具调用不灵活 | ⭐⭐⭐ 只适合检索 |
| AutoGPT | 全自动任务 | 真的完全自主 | 不可控，烧钱，容易跑偏 | ⭐⭐ 玩具级别 |
| CrewAI | 多Agent协作 | 角色分工清晰 | 学习曲线陡，中文资料少 | ⭐⭐⭐⭐ 值得学 |
| 自己写 | 生产环境 | 完全可控，能优化性能 | 开发成本高，要自己维护 | ⭐⭐⭐⭐⭐ 最终方案 |
血泪教训： 别一上来就用AutoGPT。我们最开始被它的demo视频忽悠了，觉得"哇，全自动，太牛了"。实际用起来，它会在你不知情的情况下狂调API，一天烧了$200，还把数据搞乱了。
二、实战：从0搭一个数据分析Agent
下面分享我们实际落地的案例。这个Agent能自动完成：数据提取→清洗→分析→可视化→写报告，原来运营团队2小时的工作，现在5分钟搞定。
2.1 需求背景（为什么要做这个？）
原始流程（很low但很真实）：
运营团队每天要写数据报告，流程是这样的：
- 登录数据库，手写SQL（经常语法错）
- 导出Excel到本地
- 用Excel或者pandas做分析
- 用Matplotlib画个图（丑得不行）
- 粘贴到PPT，写分析报告
痛点（运营同学的原话）：
- “每份报告平均2小时，手都点废了”
- “SQL老是写错，还要找开发帮忙”
- “图太丑了，老板每次都要说”
我们的方案： 做一个Agent，运营同学只需要说一句话：“帮我分析上季度销售数据，找出TOP 10商品”，Agent自动完成全流程。
2.2 系统架构（我们是怎么设计的）
先上架构图（这是我们生产环境的真实架构）：
用户说："分析上季度销售数据"
↓
┌─────────────────────┐
│ 意图解析模块 │ ← 识别：这是数据分析任务
│ - 时间：上季度 │
│ - 指标：销售额 │
│ - 维度：商品 │
└─────────┬───────────┘
↓
┌─────────────────────┐
│ 任务规划模块 │ ← 生成执行计划
│ 1. 连接数据库 │
│ 2. 数据清洗 │
│ 3. 按商品聚合 │
│ 4. 计算TOP 10 │
│ 5. 生成图表 │
│ 6. 写分析报告 │
└─────────┬───────────┘
↓
┌─────────────────────┐
│ 工具执行器 │ ← 一步步执行
│ - SQL Tool │
│ - Python Tool │
│ - Chart Tool │
│ - Report Tool │
└─────────┬───────────┘
↓
┌─────────────────────┐
│ 结果反思模块 │ ← 发现问题自动调整
│ - 数据量合理吗？ │
│ - 图表清晰吗？ │
│ - 结论有数据支撑？│
└─────────┬───────────┘
↓
输出报告（Word/PDF）
关键点： 每个模块都是独立的，可以单独升级。比如，后来我们发现"意图解析"不够准，就换了个更好的LLM，其他模块不用动。
2.3 核心代码（可直接用的版本）
（1）Agent主循环（核心中的核心）
from typing import List, Dict, Any
import openai
import json
class DataAnalysisAgent:
def __init__(self, db_connection, max_retries=5):
"""
初始化Agent
db_connection: 数据库连接
max_retries: 每个步骤最多重试几次
"""
self.llm = openai.ChatCompletion # 我们用GPT-4 Turbo
self.db = db_connection
self.max_retries = max_retries
# 注册工具（告诉Agent它能用什么）
self.tools = {
"sql_query": self.sql_tool,
"python_process": self.python_tool,
"generate_chart": self.chart_tool,
"generate_report": self.report_tool
}
def run(self, user_query: str) -> str:
"""
Agent主循环：规划 → 执行 → 反思 → 再执行
这是整个系统的核心逻辑。
如果看懂了这个，后面都是细节。
"""
# === 第1步：理解用户意图 ===
intent = self.parse_intent(user_query)
print(f"[调试] 识别到意图：{intent}")
# === 第2步：生成执行计划 ===
# 比如：连接数据库 → 查询数据 → 清洗 → 分析 → 可视化 → 写报告
plan = self.create_plan(intent)
print(f"[调试] 生成执行计划：{plan}")
# === 第3步：执行任务（带反思机制） ===
results = [] # 存每步的结果
context = "" # 用于多轮对话的上下文
for step in plan['steps']:
for attempt in range(self.max_retries):
try:
# 执行当前步骤（可能调用SQL、Python、画图等）
result = self.execute_step(step, results, context)
# 执行成功，记录结果
results.append({
"step": step,
"result": result,
"status": "success"
})
# 更新上下文（让后续步骤知道前面发生了啥）
context += f"\n步骤'{step}'执行结果：{result}"
print(f"[调试] 步骤'{step}'执行成功")
break # 成功就跳出重试循环
except Exception as e:
print(f"[调试] 步骤'{step}'执行失败：{e}")
# === 关键：反思为什么失败 ===
reflection = self.reflect_on_error(e, step, results, context)
print(f"[调试] 反思结果：{reflection}")
# 如果重试次数用完，放弃这个步骤
if attempt == self.max_retries - 1:
results.append({
"step": step,
"result": f"执行失败：{e}",
"status": "failed"
})
print(f"[警告] 步骤'{step}'重试{self.max_retries}次仍失败，跳过")
break
# 根据反思结果，调整步骤后重试
step = self.adjust_step(step, reflection)
print(f"[调试] 调整为：{step}")
# === 第4步：生成最终报告 ===
report = self.generate_report(results, intent)
return report
def parse_intent(self, user_query: str) -> Dict[str, Any]:
"""
用LLM解析用户意图
实际应用：我们发现直接用LLM做意图解析，准确率能到92%
比手写规则牛逼太多
"""
prompt = f"""
解析以下用户查询的意图，输出JSON格式：
用户查询：{user_query}
输出格式（严格JSON，不要有其他内容）：
{{
"task_type": "数据分析|文本生成|图像生成|...",
"time_range": "时间范围（如果提到）",
"metrics": ["指标1", "指标2", ...],
"dimensions": ["维度1", "维度2", ...],
"additional_requirements": "其他要求"
}}
"""
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
temperature=0 # 关键：设为0保证输出稳定
)
# 解析JSON（记得处理异常）
try:
intent_json = response['choices'][0]['message']['content']
return json.loads(intent_json)
except:
# 如果JSON解析失败，返回默认意图
return {
"task_type": "数据分析",
"time_range": "最近",
"metrics": [],
"dimensions": [],
"additional_requirements": ""
}
def create_plan(self, intent: Dict[str, Any]) -> Dict[str, Any]:
"""
根据意图，生成执行计划
实际应用：这里我们用Few-shot（给几个历史成功案例）
让LLM学会"正确的思考方式"
"""
# 把intent转成字符串，塞给LLM
intent_str = json.dumps(intent, ensure_ascii=False)
prompt = f"""
根据以下意图，生成数据分析的执行计划（步骤列表）：
意图：{intent_str}
要求：
1. 步骤应该具体、可执行（不要写"分析数据"这种废话）
2. 每步只做一件事
3. 考虑数据依赖（后面的步骤可能依赖前面的结果）
输出格式（严格JSON）：
{{
"steps": [
"步骤1描述",
"步骤2描述",
...
]
}}
只输出JSON，不要有其他内容。
"""
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
plan_json = response['choices'][0]['message']['content']
return json.loads(plan_json)
def execute_step(self, step: str, previous_results: List[Dict], context: str) -> str:
"""
执行单个步骤（可能调用工具）
这是最核心的方法：让LLM判断"这一步该用哪个工具"
"""
# 构造prompt，让LLM决定调用哪个工具
prompt = f"""
当前步骤：{step}
已执行结果：
{json.dumps(previous_results, ensure_ascii=False, indent=2)}
可用工具：
- sql_query: 执行SQL查询（输入：SQL语句）
- python_process: 用Python处理数据（输入：pandas代码）
- generate_chart: 生成图表（输入：图表类型和参数）
- generate_report: 生成报告（输入：报告要求）
请判断该步骤应该调用哪个工具，以及输入是什么。
输出格式（严格JSON）：
{{
"tool": "工具名",
"input": "工具输入"
}}
只输出JSON，不要有其他内容。
"""
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
# 解析LLM的输出，得到"用哪个工具"和"输入是什么"
decision = json.loads(response['choices'][0]['message']['content'])
tool_name = decision['tool']
tool_input = decision['input']
# 调用对应的工具
if tool_name in self.tools:
result = self.tools[tool_name](tool_input)
return result
else:
raise ValueError(f"未知工具：{tool_name}，可用的工具：{list(self.tools.keys())}")
def reflect_on_error(self, error, step, previous_results, context):
"""
反思错误，调整策略
实际应用：这是Agent的"智能"所在
发现错了 → 分析原因 → 调整方案 → 再试
"""
prompt = f"""
执行步骤时出错，请分析原因并给出调整方案：
步骤：{step}
错误类型：{type(error).__name__}
错误信息：{str(error)}
已执行结果：{json.dumps(previous_results, ensure_ascii=False)}
请分析：
1. 错误原因是什么？（SQL语法错？数据为空？网络超时？）
2. 应该如何调整步骤？
3. 是否需要更换工具？
输出格式（严格JSON）：
{{
"error_reason": "错误原因",
"adjustment": "调整方案",
"new_step": "调整后的步骤描述（如果不需要调整，保持原步骤）"
}}
只输出JSON，不要有其他内容。
"""
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return json.loads(response['choices'][0]['message']['content'])
# ========== 工具定义（Agent能干什么） ==========
def sql_tool(self, sql_query: str) -> str:
"""
执行SQL查询工具
安全注意：生产环境一定要限制SQL类型（只允许SELECT）
"""
import pandas as pd
# === 安全检查：只允许SELECT ===
if not sql_query.strip().lower().startswith('select'):
raise ValueError("只允许SELECT查询，禁止INSERT/UPDATE/DELETE")
try:
# 执行SQL，返回DataFrame
df = pd.read_sql(sql_query, self.db)
return f"查询成功，返回{len(df)}行数据。\n前5行：\n{df.head().to_string()}"
except Exception as e:
raise Exception(f"SQL执行失败：{e}")
def python_tool(self, code: str) -> str:
"""
执行Python代码工具（用pandas处理数据）
安全注意：生产环境应该用沙箱（如Docker容器）
这里简化实现，直接exec（不安全！）
"""
import pandas as pd
import sys
from io import StringIO
try:
# 重定向stdout，捕获print的输出
old_stdout = sys.stdout
sys.stdout = StringIO()
# 执行代码（危险！生产环境千万别这么干）
exec(code, {'pd': pd, 'df': self.current_df})
output = sys.stdout.getvalue()
sys.stdout = old_stdout
return output if output else "代码执行成功（无输出）"
except Exception as e:
sys.stdout = old_stdout
raise Exception(f"Python执行失败：{e}")
def chart_tool(self, chart_config: str) -> str:
"""生成图表工具（简化实现）"""
import matplotlib.pyplot as plt
import json
config = json.loads(chart_config)
chart_type = config.get('type', 'line')
plt.figure(figsize=(10, 6))
if chart_type == 'bar':
plt.bar(config['x'], config['y'])
elif chart_type == 'line':
plt.plot(config['x'], config['y'])
plt.title(config.get('title', 'Chart'))
plt.xlabel(config.get('xlabel', ''))
plt.ylabel(config.get('ylabel', ''))
# 保存到临时文件
import tempfile
with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as f:
plt.savefig(f.name)
plt.close()
return f"图表已生成：{f.name}"
def report_tool(self, report_req: str) -> str:
"""生成报告工具（简化实现）"""
return f"报告生成工具被调用，要求：{report_req}"
def generate_report(self, results: List[Dict], intent: Dict) -> str:
"""
根据执行结果，生成最终报告
实际应用：这里会让LLM根据数据，自动写分析报告
包括：摘要、数据、分析、结论、建议
"""
prompt = f"""
根据以下执行结果，生成一份数据分析报告：
用户需求：{json.dumps(intent, ensure_ascii=False)}
执行结果：
{json.dumps(results, ensure_ascii=False, indent=2)}
报告要求：
1. 结构清晰（摘要、数据、分析、结论）
2. 语言专业但易懂（不要堆砌术语）
3. 给出可操作的建议（不要只说"有待提高"）
请生成完整报告。
"""
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
temperature=0.3 # 报告可以有点创造性
)
return response['choices'][0]['message']['content']
# ========== 如何使用 ==========
if __name__ == "__main__":
# 假设已经有数据库连接（这里用sqlite举例）
import sqlite3
db = sqlite3.connect('sales.db')
# 创建Agent
agent = DataAnalysisAgent(db)
# 执行任务（说人话就行）
result = agent.run("帮我分析上季度销售数据，找出TOP 10商品")
print("\n========== 最终报告 ==========")
print(result)
上面的代码能直接跑吗？ 差不多，但还有些细节要补（比如self.current_df要在sql_tool执行后赋值）。完整可运行版本在我的GitHub（如果觉得文章有用，点个赞我再整理上传 -_-）。
（2）关键优化：用Few-shot提升规划质量
我们发现，直接让LLM生成执行计划，准确率只有65%左右。后来加了几个历史成功案例（Few-shot），准确率直接飙到92%。
方法： 在create_plan的prompt里，加入这样的案例：
FEW_SHOT_EXAMPLES = """
以下是历史成功案例，请学习其规划思路：
=== 案例1 ===
用户问题："分析上季度销售数据，找出TOP商品"
规划步骤：
1. 先确认时间范围（"上季度"具体是哪几个月？→ 2025年Q4：10/11/12月）
2. 连接sales数据库，查询order_items表
3. 按month聚合，验证数据完整性
4. 发现12月数据缺失 → 补充查询
5. 按商品(product_id)维度聚合销售额
6. 计算TOP 10，同时计算环比增长
7. 生成柱状图（TOP 10商品销售额）
8. 撰写分析结论（哪些商品表现好？为什么？）
=== 案例2 ===
用户问题："对比北京和上海的用户增长情况"
规划步骤：
1. 确认时间范围（默认最近3个月）
2. 查询users表，按city='北京'和city='上海'分别统计
3. 按月份聚合，计算月度新增用户
4. 发现上海2月数据异常低 → 检查是否有数据缺失
5. 生成双折线图（北京vs上海月度增长对比）
6. 计算增长率，分析差异原因
7. 给出运营建议
请参考以上案例的规划思路，为当前问题制定执行计划。
"""
**为什么Few-shot有效？** LLM是通过"例子"学习的。你给它几个高质量案例，它就能模仿那种"思考方式"。
**效果：** 规划准确率从65% → 92%，大幅减少重试次数。
### 2.4 生产环境踩坑实录
理论说完，该讲实战了。以下都是我们真真切切踩过的坑。
#### 坑1：LLM输出不稳定（有时候生成的SQL是错的）
**现象：** 同样的输入，有时候生成的SQL正确，有时候语法错误，有时候逻辑错误。
**原因分析：**
1. LLM本身是概率模型，`temperature>0`时输出会随机
2. 复杂任务需要多步推理，一步错步步错
3. 没有统一的中间格式，难以验证
**解决方案（我们试了3个）：**
**方案1：用JSON Schema强制输出格式**
```python
def generate_sql_with_schema(self, user_query: str) -> dict:
"""生成SQL，强制JSON格式输出"""
prompt = f"""
根据以下问题生成SQL查询：
{user_query}
输出必须严格符合以下JSON格式（不要有其他内容）：
{{
"sql": "SELECT ... FROM ... WHERE ...",
"explanation": "查询说明",
"potential_issues": ["可能的问题1", "可能的问题2"]
}}
"""
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
temperature=0, # 关键：设为0保证稳定性
response_format={ "type": "json_object" } # 强制JSON输出（新功能）
)
result = json.loads(response['choices'][0]['message']['content'])
sql = result['sql']
# 语法检查（双重保险）
if not self.validate_sql_syntax(sql):
# 自动修复：提取SQL并修正
sql = self.fix_sql_syntax(sql)
result['sql'] = sql
return result
方案2：增加"语法检查 + 自动修复"环节
def validate_sql_syntax(self, sql: str) -> bool:
"""用sqlparse验证SQL语法"""
import sqlparse
try:
parsed = sqlparse.parse(sql)
if not parsed:
return False
# 检查是否包含禁止的操作（安全！）
forbidden = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'TRUNCATE']
for token in parsed[0].tokens:
if token.ttype is sqlparse.tokens.Keyword and str(token).upper() in forbidden:
return False
return True
except:
return False
def fix_sql_syntax(self, sql: str) -> str:
"""尝试自动修复SQL语法错误"""
# 去除可能的markdown代码块标记
sql = sql.replace('```sql', '').replace('```', '')
# 去除末尾的分号（如果需要）
sql = sql.strip().rstrip(';')
return sql
方案3：对关键步骤设置temperature=0
# 生成SQL、Python代码等关键步骤，必须用temperature=0
def execute_critical_step(self, step):
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[...],
temperature=0 # 保证输出稳定（但可能缺乏创造性）
)
return response
# 生成报告、总结等创造性任务，可以用temperature=0.3~0.7
def generate_report(self, results):
response = self.llm.create(
model="gpt-4-turbo-preview",
messages=[...],
temperature=0.5 # 允许一定创造性
)
return response
效果： SQL生成成功率从78% → 97%，大幅减少了重试次数。
坑2：成本失控（一个任务烧了$50）
现象： 一个复杂任务可能调用LLM几百次，费用爆炸（单次任务$5~$10，甚至有次烧了$50）。
原因分析：
- 每次工具调用都要请求LLM（判断用哪个工具+生成输入）
- 反思-重试机制可能导致多轮对话
- 没有提前终止机制（明明已经得到答案还继续调）
解决方案：
方案1：简单任务用小型模型，复杂任务才用GPT-4
def select_model(self, task_complexity: int) -> str:
"""
根据任务复杂度选择模型
task_complexity: 1~10，分数越高越复杂
"""
if task_complexity <= 3:
return "gpt-3.5-turbo" # 简单任务，便宜快速
elif task_complexity <= 7:
return "gpt-4-turbo-preview" # 中等任务
else:
return "gpt-4" # 复杂任务，效果好但贵
def estimate_complexity(self, user_query: str) -> int:
"""预估任务复杂度（1~10）"""
# 简单启发式规则
complexity = 1
if '分析' in user_query or '对比' in user_query:
complexity += 2
if '可视化' in user_query or '图表' in user_query:
complexity += 1
if '报告' in user_query or '总结' in user_query:
complexity += 1
# 也可以让LLM帮忙评估（用便宜的模型）
prompt = f"评估以下任务的复杂度（1~10分）：{user_query}\n只输出数字。"
response = self.llm.create(
model="gpt-3.5-turbo", # 用便宜的模型
messages=[{"role": "user", "content": prompt}],
temperature=0
)
try:
llm_score = int(response['choices'][0]['message']['content'].strip())
complexity = max(complexity, llm_score)
except:
pass
return min(complexity, 10)
方案2：实现"提前终止"机制
def should_stop_early(self, intermediate_results: List[Dict], user_query: str) -> tuple:
"""
判断是否可以提前终止
实际应用：有时候前3步就已经能回答问题了
没必要非得把所有步骤都走完
"""
prompt = f"""
基于目前的中间结果，能否回答用户问题？
用户问题：{user_query}
中间结果：{json.dumps(intermediate_results, ensure_ascii=False)}
如果能回答，请直接给出答案，格式：
{{
"can_answer": true,
"answer": "答案内容"
}}
如果不能，请继续：
{{
"can_answer": false,
"next_step_suggestion": "建议的下一步"
}}
只输出JSON，不要有其他内容。
"""
response = self.llm.create(
model="gpt-3.5-turbo", # 用便宜的模型做判断
messages=[{"role": "user", "content": prompt}],
temperature=0
)
result = json.loads(response['choices'][0]['message']['content'])
if result['can_answer']:
return (True, result['answer'])
else:
return (False, None)
# 在主循环中调用
def run(self, user_query: str) -> str:
# ... 前面的代码 ...
for step in plan['steps']:
# 执行步骤
result = self.execute_step(step, results, context)
results.append(result)
# 检查是否可以提前终止（省时间与金钱！）
should_stop, answer = self.should_stop_early(results, user_query)
if should_stop:
print("[Early Stop] 已获得足够信息，提前终止")
return answer
# 正常执行完所有步骤
report = self.generate_report(results, intent)
return report
方案3：用缓存减少重复调用
from sentence_transformers import SentenceTransformer, util
class SmartCache:
"""智能缓存：语义相似的问题复用结果"""
def __init__(self):
# 用多语言模型（支持中文）
self.model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
self.cache = {} # {query: (embedding, result)}
def get(self, query: str, threshold: float = 0.92) -> str:
"""
查询缓存（语义相似度）
threshold: 相似度阈值，>0.92才复用
"""
query_embedding = self.model.encode(query, convert_to_tensor=True)
for cached_query, (cached_embedding, result) in self.cache.items():
similarity = util.cos_sim(query_embedding, cached_embedding).item()
if similarity > threshold:
print(f"[Cache] 命中缓存：{cached_query} (相似度={similarity:.3f})")
return result
return None
def set(self, query: str, result: str):
"""写入缓存"""
query_embedding = self.model.encode(query, convert_to_tensor=True)
self.cache[query] = (query_embedding, result)
# 限制缓存大小（最多100条，防止内存爆炸）
if len(self.cache) > 100:
# 删除最早的一条（简化实现）
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
# 在Agent中使用缓存
class DataAnalysisAgent:
def __init__(self, db_connection):
# ... 其他初始化 ...
self.cache = SmartCache()
def run(self, user_query: str) -> str:
# 先查缓存（命中就直接返回，省钱！）
cached_result = self.cache.get(user_query)
if cached_result:
return cached_result
# 正常执行
result = self._run_without_cache(user_query)
# 写入缓存
self.cache.set(user_query, result)
return result
效果： 单次任务平均成本从$5 → $0.8，缓存命中率约30%。
三、效果评估与持续优化
3.1 我们怎么评估Agent的好坏？
我们定义了多层级的评估体系：
| 层级 | 指标 | 怎么算 | 目标值 | 我们目前的值 |
|---|---|---|---|---|
| 任务完成度 | 成功完成率 | 成功任务数/总任务数 | >90% | 94% |
| 结果质量 | 人工评分 | 1~5分（3个评审取平均） | 平均>4.0 | 4.2 |
| 性能 | 平均响应时间 | 所有任务响应时间平均 | <30秒 | 22秒 |
| 成本 | 单次任务平均成本 | 总成本/总任务数 | <$0.5 | $0.8（还在优化） |
| 用户满意度 | NPS评分 | 推荐者%-贬损者% | >50 | 62 |
人工评分标准（1~5分）：
- 5分：结果完全符合预期，数据准确，结论有洞察，可直接使用
- 4分：结果基本符合预期，有小瑕疵但不影响使用
- 3分：结果部分符合预期，需要人工修正
- 2分：结果不符合预期，但有参考价值
- 1分：结果完全错误或无意义
3.2 持续优化策略（我们怎么让Agent越来越聪明？）
策略1：失败案例复盘制度
流程（每周五下午，团队集中复盘）：
- 找出本周失败案例（为什么这10个任务没做成？）
- 分析失败原因（规划错误/工具选择错误/LLM输出错误/数据源问题…）
- 更新Few-shot示例库和提示词
- 记录到知识库，避免重复踩坑
def weekly_review(self):
"""每周复盘失败案例"""
import os
from datetime import datetime, timedelta
# 获取本周失败案例
today = datetime.now()
week_ago = today - timedelta(days=7)
failed_cases = self.get_failed_cases(start_date=week_ago, end_date=today)
print(f"=== 本周失败案例复盘（共{len(failed_cases)}个）===")
for case in failed_cases:
print(f"\n--- 案例ID: {case['id']} ---")
print(f"用户问题：{case['user_query']}")
print(f"失败原因：{case['failure_reason']}")
# 分析失败原因（用LLM）
root_cause = self.analyze_failure(case)
print(f"根本原因：{root_cause}")
# 根据根本原因，更新对应的内容
if root_cause == "规划错误":
self.update_few_shots('planning', case)
print("✓ 已更新规划Few-shot示例")
elif root_cause == "工具选择错误":
self.update_tool_description(case['failed_tool'])
print(f"✓ 已更新工具{case['failed_tool']}的描述")
elif root_cause == "LLM输出格式错误":
self.update_output_format_requirement(case)
print("✓ 已更新输出格式要求")
# 记录到知识库（避免以后再踩）
self.knowledge_base.add(case)
print()
策略2：A/B测试不同提示词
对于新上线的提示词优化，先做小流量A/B测试：
import random
def ab_test_prompt(self, prompt_A: str, prompt_B: str, test_queries: List[str], traffic_split: float = 0.5):
"""
A/B测试提示词
prompt_A: 原提示词
prompt_B: 新提示词
test_queries: 测试查询列表
traffic_split: 流量分配（A组比例）
"""
results = {'A': [], 'B': []}
for user_query in test_queries:
if random.random() < traffic_split:
# A组：用原提示词
result = self.run_with_prompt(prompt_A, user_query)
results['A'].append({
"query": user_query,
"result": result,
"score": self.evaluate_result_quality(result)
})
else:
# B组：用新提示词
result = self.run_with_prompt(prompt_B, user_query)
results['B'].append({
"query": user_query,
"result": result,
"score": self.evaluate_result_quality(result)
})
# 对比效果
avg_score_A = sum([r['score'] for r in results['A']]) / len(results['A'])
avg_score_B = sum([r['score'] for r in results['B']]) / len(results['B'])
print(f"=== A/B测试结果 ===")
print(f"A组（原提示词）平均得分：{avg_score_A:.2f}")
print(f"B组（新提示词）平均得分：{avg_score_B:.2f}")
if avg_score_B > avg_score_A:
print(f"✓ B组胜出，新提示词效果更好（提升{avg_score_B - avg_score_A:.2f}分）")
return 'B'
else:
print(f"✗ B组未胜出，继续用原提示词（差距{avg_score_A - avg_score_B:.2f}分）")
return 'A'
四、总结（干货提炼）
4.1 核心要点（建议收藏）
- AI Agent ≠ 简单调用LLM，需要三大能力：规划、工具调用、反思
- 框架选择要权衡：快速原型用LangChain，生产环境建议自研核心流程
- 提示词工程是核心：Few-shot示例、工具描述、输出格式定义，一个都不能少
- 生产环境必须考虑：成本控制、错误处理、性能优化
- 持续优化是必修课：失败复盘、A/B测试、知识库积累
4.2 实战建议（避坑指南）
✅ 推荐做法：
- 用JSON Schema强制LLM输出格式（减少解析错误）
- 关键步骤设置
temperature=0保证稳定性 - 实现提前终止机制降低成本（别做无用功）
- 用语义缓存复用相似查询结果（我们命中率30%）
- 建立失败案例复盘制度（每周五下午，坚持了6个月）
❌ 常见误区（我们踩过的坑）：
- 以为调用LLM就是Agent（缺少规划+工具，只是个问答机）
- 没有错误处理，工具调用失败就崩溃（生产环境大忌）
- 所有步骤都用最贵的模型（成本失控，老板找你谈话）
- 没有评估体系，不知道效果好坏（瞎优化）
4.3 未来方向（我们在研究的）
- 多模态Agent：支持图像、音频、视频理解（如：分析图表、听录音整理会议纪要）
- 多Agent协作：不同专长的Agent协同完成复杂任务（如：一个负责数据抓取，一个负责分析，一个负责写报告）
- 自主学习：Agent能从失败中学习，自动优化策略和提示词（不用人工干预）
- 个性化适配：根据用户反馈调整行为风格（如：老板要结论，组员要细节）
参考资料（都是我们实际用过的）
- LangChain官方文档 - 快速原型必备
- ReAct论文 - Agent的经典范式，必读
- Toolformer论文 - 工具调用学习，很硬核
- AutoGPT项目 - 可以玩玩，但别用在生产环境
- CrewAI官方文档 - 多Agent协作，值得学
如果本文帮你少踩了坑，请点赞 + 收藏 + 关注三连！
讨论题：
- 你们团队在做AI Agent时踩过哪些坑？欢迎评论区分享！
- 你觉得AI Agent最重要的是哪个能力？规划？工具调用？还是反思？
- 有没有比LangChain更好的Agent框架推荐？
期待你的分享和讨论！
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
12
0- 0
已为社区贡献26条内容
所有评论(0)