共计 2867 个字符，预计需要花费 8 分钟才能阅读完成。
背景痛点分析
最近在做一个客服 Agent 项目时，踩了不少坑。总结下来当前 LLM Agent 开发主要存在三大问题：
-
状态管理混乱：很多实现把对话历史直接拼接成字符串传给 LLM，当对话轮次超过 10 轮后，不仅 token 开销爆炸，关键上下文还容易丢失。上周我们就遇到用户问了 5 个问题后，Agent 完全忘记最初需求的尴尬情况
-
工具调用脆弱：调用外部 API 时没有超时控制，有一次天气查询接口挂掉导致整个会话卡死 30 秒。更糟的是有些开发者会无脑重试，把错误次数也计入 prompt，结果 LLM 开始胡言乱语说 ” 系统正在第 5 次重试 …”
-
记忆模块简陋：所有对话记忆存在一个列表里，既没区分短期记忆（当前对话）和长期记忆（用户画像），也没有检索优化。测试时发现 Agent 总是引用 3 天前的旧地址，而用户明明刚更新过收货信息
模块化设计方案
经过几次迭代，我们总结出四层架构（附示意图）：
graph TD
A[Interface Layer] --> B[Orchestration Layer]
B --> C[Memory Layer]
B --> D[Tools Layer]关键技术决策
- 记忆分层：
- 短期记忆：固定窗口大小的对话历史（最近 10 轮）
- 长期记忆：向量数据库存储关键事实（用户偏好 / 业务规则）
-
元记忆：记录工具调用日志等系统级信息
-
工具熔断设计：当工具连续失败 3 次时，自动将其禁用 5 分钟，并通知运维系统。这里借鉴了 Circuit Breaker 模式：
class CircuitBreaker:
def __init__(self, max_fails=3, cool_down=300):
self._fails = 0
self._last_fail = 0
async def call(self, tool_func, *args):
if time.time() - self._last_fail < self.cool_down:
raise ToolTemporarilyDisabled()
try:
return await tool_func(*args)
except Exception as e:
self._fails += 1
if self._fails >= self.max_fails:
self._last_fail = time.time()
raise核心代码实现
带类型约束的 Agent 基类
from typing import Protocol, runtime_checkable
@runtime_checkable
class Tool(Protocol):
name: str
description: str
async def __call__(self, input: str) -> str: ...
class AgentCore:
def __init__(self, tools: list[Tool]):
self._validate_tools(tools) # 检查工具是否符合协议
self.memory = VectorMemory()
def _validate_tools(self, tools):
for tool in tools:
if not isinstance(tool, Tool):
raise TypeError(f"工具 {tool.name} 未实现 Tool 协议")记忆检索优化
class VectorMemory:
def __init__(self):
self._storage = [] # (vector, metadata)元组列表
self._cache = LRUCache(maxsize=1000)
def add_memory(self, text: str, importance: float):
"""
参数 importance 控制记忆保留优先级
0.5 为默认值，>0.8 的记忆会持久化到数据库
"""
vector = get_embedding(text)
self._storage.append((vector, {"text": text, "imp": importance}))
def retrieve(self, query: str, top_k=3) -> list[str]:
"""
使用余弦相似度过滤低相关性记忆
对近期记忆给予时间衰减补偿
"""
query_vec = get_embedding(query)
scores = []
for vec, meta in self._storage:
sim = cosine_similarity(query_vec, vec)
time_decay = 0.9 ** meta["age"] # 每天衰减 10%
scores.append((sim * time_decay * meta["imp"], meta["text"]))
return [text for _, text in sorted(scores, reverse=True)[:top_k]]生产环境验证
压力测试结果
使用 locust 模拟 100 并发用户时的表现：
| 指标 | 数值 |
|---|---|
| 平均响应延迟 | 320ms |
| 95 分位延迟 | 890ms |
| 错误率 | 0.12% |
关键优化点：
– 对 LLM 的调用实现了请求批处理
– 向量检索改用 FAISS 加速
– 工具调用线程池限制最大并发数
安全防护措施
- 输入过滤：
def sanitize_input(text: str) -> str: # 移除可能破坏 prompt 的特殊符号 return text.translate(str.maketrans({"<": "",">":"", "{": "","}":""})) - 权限控制：每个工具注册时需要声明 required_scopes
避坑经验分享
记忆污染案例：
曾经有用户问 ” 忘记密码怎么办 ”，Agent 回答步骤时，不小心把测试用的临时密码 ”123456″ 也包含在记忆里。之后所有密码相关提问都会引用这个错误值。现在我们采用以下策略：
- 敏感信息检测（正则匹配手机号 / 密码等）
- 重要事实需要用户二次确认才会存入长期记忆
- 定期运行记忆清洗任务
工具权限清单示例：
- name: process_refund
scopes: [finance]
rate_limit: 5/ 分钟
- name: query_weather
scopes: [public]
timeout: 2 秒延伸思考
遇到一个有趣问题：当多个 Agent 需要协作时（比如销售 Agent 和技术支持 Agent 同时服务客户），如何解决这些场景：
1. 两个 Agent 同时调用支付接口
2. 对同一问题给出矛盾回答
3. 资源竞争（如同时请求语音合成服务）
我们搭建了一个测试沙盒，包含三个预配置 Agent：
POST /api/chat/multi-agent
{"agents": ["sales", "support", "translator"],
"message": "这个价格能否包含安装服务？"
}欢迎读者尝试后分享解决方案，目前我们采用的冲突解决策略是：
– 操作冲突：分布式锁 + 事务日志
– 回答冲突：基于 Agent 角色权重的投票机制
– 资源竞争：优先级队列 + 预分配配额
这套架构已在电商客服场景运行 6 个月，成功将平均问题解决轮次从 4.3 降低到 2.1。最大的收获是：Agent 系统不是越复杂越好，清晰的边界定义比算法优化更重要。
