共计 3340 个字符，预计需要花费 9 分钟才能阅读完成。
技术背景：为什么需要 AI Agent
在客服自动化、智能流程处理等场景中，传统规则引擎依赖硬编码逻辑，遇到复杂需求时维护成本指数级上升。比如处理 ” 我要退上个月买的手机但发票丢了 ” 这样的长尾请求，需要编写数百条 if-else 规则。而基于 LLM 的 AI Agent 通过理解自然语言意图，配合外部工具调用（查订单、验证身份等），能优雅解决这类问题。
架构选型对比
| 技术方案 | 开发效率 | 灵活性 | 学习成本 | 适用场景 |
|---|---|---|---|---|
| LangChain | 高 | 中 | 低 | 快速原型开发 |
| Semantic Kernel | 中 | 高 | 高 | 企业级复杂系统 |
| 原生 API 调用 | 低 | 极高 | 中 | 深度定制需求 |
实践建议 ：中小项目建议从 LangChain 开始，当需要精细控制 LLM 输入输出时再切到原生 API。
核心实现
带重试机制的 Tool Agent
import backoff
import openai
@backoff.on_exception(backoff.expo,
(openai.error.APIError,
openai.error.Timeout),
max_tries=3)
def call_llm_with_tools(messages, tools):
"""
:param messages: 对话历史列表
:param tools: 可调用工具描述 (OpenAI 格式)
:return: LLM 响应 (可能包含工具调用请求)
"""
return openai.ChatCompletion.create(
model="gpt-3.5-turbo-0613",
messages=messages,
tools=tools,
tool_choice="auto"
)关键点说明：
- 使用 backoff 实现指数退避重试
- tool_choice=”auto” 让模型自主决定是否调用工具
- 工具描述需符合 OpenAI 规范（示例见下文）
对话状态管理方案
无状态会话 （适合简单场景）：
def handle_message(session_id, user_input):
# 从 Redis 获取历史记录
history = redis.get(f"chat:{session_id}") or []
# 追加新消息并截断（防止超长）history.append({"role": "user", "content": user_input})
history = history[-10:] # 保留最近 10 轮
# 调用 LLM 并保存结果
response = call_llm(history)
redis.setex(f"chat:{session_id}", 3600, history)
return response有状态会话 （复杂业务流程）：
class OrderTrackingAgent:
def __init__(self, session_id):
self.state = {
"step": "ask_order_number",
"collected_data": {},
"history": []}
def process_input(self, user_input):
if self.state["step"] == "ask_order_number":
self.state["collected_data"]["order_no"] = user_input
self.state["step"] = "verify_payment"
return "请提供付款手机尾号后 4 位"
# ... 其他状态处理逻辑 性能优化实战
延迟优化三板斧
-
请求批处理 ：将多个用户查询合并为一个 API 调用
# 使用 OpenAI 的 batch 接口 responses = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[...], # 多个对话上下文组成的列表 ) -
结果缓存 ：
from diskcache import Cache @cache.memoize(expire=300) def get_cached_response(prompt): return call_llm([{"role": "user", "content": prompt}]) -
流式响应 ：
for chunk in openai.ChatCompletion.create( model="gpt-4", messages=messages, stream=True ): print(chunk.choices[0].delta.get("content", ""), end="")
速率限制处理
def adaptive_rate_limiter():
"""智能调节请求间隔的动态限流器"""
last_call = time.time()
min_interval = 0.1 # 初始 100ms
def decorator(func):
@wraps(func)
def wrapped(*args, **kwargs):
nonlocal min_interval
elapsed = time.time() - last_call
if elapsed < min_interval:
time.sleep(min_interval - elapsed)
try:
result = func(*args, **kwargs)
min_interval = max(0.05, min_interval * 0.9) # 成功时加快
return result
except openai.error.RateLimitError:
min_interval = min(2.0, min_interval * 1.5) # 失败时减速
raise
return wrapped
return decorator必须掌握的避坑技巧
敏感信息过滤
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
def sanitize_input(text):
# 检测信用卡号、手机号等
results = analyzer.analyze(text=text, language="zh")
# 替换为 [REDACTED]
anonymized = anonymizer.anonymize(
text=text,
analyzer_results=results
)
return anonymized.text异步任务队列选型
| 特性 | Celery | Dramatiq |
|---|---|---|
| 吞吐量 | 中 | 高 |
| 内存占用 | 高 | 低 |
| 定时任务 | 支持 | 需插件 |
| 监控界面 | 内置 Flower | 需第三方 |
推荐选择 ：
– 需要复杂工作流用 Celery
– 追求极致性能选 Dramatiq
可观测性设计
建议采集这些核心指标：
# Prometheus 格式的指标示例
from prometheus_client import Counter, Histogram
LLM_CALLS = Counter('llm_calls_total', 'Total LLM API calls')
TOKEN_USAGE = Histogram('llm_tokens_used', 'Token consumption per call')
def instrumented_llm_call(*args, **kwargs):
LLM_CALLS.inc()
start = time.time()
response = original_llm_call(*args, **kwargs)
duration = time.time() - start
TOKEN_USAGE.observe(response.usage.total_tokens)
return response 监控看板应包含：
– 每分钟请求量
– 平均响应延迟 (P99/P95)
– Token 消耗趋势
– 错误类型分布
总结与演进方向
经过多个生产项目验证，稳定的 AI Agent 系统需要：1) 完善的错误恢复机制 2) 精准的成本控制 3) 可解释的决策过程。下一步可以探索：
- 用 LlamaIndex 实现长期记忆
- 通过微调优化特定领域表现
- 结合强化学习进行对话策略优化
开发过程中最深的体会是：不要过度追求完美的一次性设计，而应该建立快速迭代的验证机制。每个新功能都先用最小原型验证效果，再逐步完善可靠性设计。
正文完
