Agent开发10个常见陷阱及避免方法（血泪总结）

系列：问题解决方案库（2/5）
难度：⭐⭐⭐ 中级
预计阅读时间：12分钟
适用人群：正在开发Agent的开发者
经验来源：实际项目中的踩坑记录

引言

开发AI Agent听起来很酷，但实际操作中会遇到很多意想不到的问题。

经过多个Agent项目的实战，我总结了10个最常见的陷阱。

如果你正在或准备开发Agent，这篇文章能帮你避开这些坑！

陷阱1：工具定义不规范

❌ 错误做法

from langchain.tools import Tool # 工具描述太简单 tool = Tool( name="search", func=search_web, description="搜索" # ← 太模糊 )

🔍 问题分析

Agent依赖工具描述来决定：

• 何时调用这个工具
• 传入什么参数
• 期望得到什么结果

描述不清会导致：

• Agent不调用工具
• 传入错误参数
• 无限循环调用

✅ 正确做法

from langchain.tools import Tool from typing import Optional def search_web(query: str, num_results: int = 5) -> str: """ 从互联网搜索相关信息 Args: query: 搜索关键词，应该简洁明确 num_results: 返回结果数量，默认5条 Returns: 搜索结果摘要，每条包含标题和简介 """ # 实现搜索逻辑 results = perform_search(query, num_results) return format_results(results) tool = Tool( name="WebSearch", func=search_web, description=( "当需要获取实时信息、新闻、或知识库中没有的内容时使用。" "输入应为简洁的搜索关键词，如'Python最新版本'。" "返回最多5条相关结果的摘要。" "注意：不要用于数学计算或逻辑推理。" ) )

最佳实践：

• 工具名称使用驼峰命名（WebSearch而非web_search）
• 描述包含：使用时机、输入格式、输出格式、限制条件
• 添加类型提示和文档字符串
• 提供使用示例

陷阱2：Prompt设计不当

❌ 错误做法

prompt = """ 你是一个助手。回答问题。 """

🔍 问题分析

• 角色定义模糊
• 没有行为约束
• 缺少输出格式要求
• 容易受到注入攻击

✅ 正确做法

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder system_prompt = """ 你是一个专业的投资研究助手，专注于A股市场分析。 【能力范围】 - 分析股票基本面数据 - 解读财务报表 - 提供投资建议（仅供参考） 【行为约束】 - 不提供具体的买卖建议 - 不预测短期股价走势 - 所有分析基于公开数据 - 遇到不确定信息时明确说明 【输出格式】 1. 先给出结论（1-2句话） 2. 再列出关键数据支撑 3. 最后说明风险提示 【注意事项】 - 使用专业但易懂的语言 - 数据要注明来源和时间 - 保持客观中立 """ prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), MessagesPlaceholder(variable_name="history"), ("human", "{question}"), ])

Prompt设计原则：

1. 明确角色：你是谁，做什么的
2. 划定边界：能做什么，不能做什么
3. 规范输出：格式、风格、长度
4. 防御注入：防止用户绕过限制

陷阱3：错误处理缺失

❌ 错误做法

def get_stock_price(symbol: str) -> float: """获取股票价格""" response = requests.get(f"https://api.example.com/price/{symbol}") return response.json()["price"] # ← 没有任何错误处理

🔍 问题分析

• API可能超时
• 返回格式可能变化
• 网络可能中断
• Agent会崩溃或返回错误信息给用户

✅ 正确做法

from typing import Optional import requests from requests.exceptions import RequestException, Timeout def get_stock_price(symbol: str, timeout: int = 10) -> Optional[str]: """ 获取股票最新价格 Args: symbol: 股票代码，如 '000001.SZ' timeout: 超时时间（秒） Returns: 价格字符串，失败时返回错误信息 """ try: response = requests.get( f"https://api.example.com/price/{symbol}", timeout=timeout ) response.raise_for_status() data = response.json() # 验证返回数据 if "price" not in data: return f"错误：API返回格式异常，缺少price字段" price = data["price"] return f"{symbol} 最新价格：{price}元" except Timeout: return f"错误：请求超时，请稍后重试" except RequestException as e: return f"错误：网络请求失败 - {str(e)}" except Exception as e: return f"错误：未知错误 - {str(e)}" # 在Agent中使用 tool = Tool( name="StockPrice", func=get_stock_price, description="获取股票最新价格。输入股票代码，返回价格或错误信息。" )

错误处理要点：

• 捕获所有可能的异常
• 返回用户友好的错误信息
• 记录详细日志（用于调试）
• 设置合理的超时时间
• 考虑重试机制

陷阱4：记忆管理混乱

❌ 错误做法

from langchain.memory import ConversationBufferMemory # 每次创建新的Memory对象 def chat(question: str): memory = ConversationBufferMemory() # ← 每次都新建 chain = ConversationChain(llm=llm, memory=memory) return chain.run(question) # 结果：每轮对话都是独立的，没有上下文

🔍 问题分析

• Memory对象生命周期错误
• 多用户会话混淆
• 内存泄漏（长期运行）
• 无法持久化

✅ 正确做法

from langchain.memory import ConversationBufferWindowMemory from langchain.chains import ConversationChain class AgentSession: """Agent会话管理器""" def __init__(self, session_id: str, max_history: int = 10): self.session_id = session_id self.memory = ConversationBufferWindowMemory( k=max_history, # 只保留最近10轮 return_messages=True ) self.chain = ConversationChain( llm=llm, memory=self.memory, verbose=False ) def chat(self, question: str) -> str: """处理一轮对话""" response = self.chain.run(input=question) return response def clear_memory(self): """清空记忆""" self.memory.clear() def get_summary(self) -> str: """获取对话摘要""" return self.memory.buffer # 使用示例 session = AgentSession(session_id="user_123") response1 = session.chat("我叫小明") response2 = session.chat("我今年25岁") response3 = session.chat("我叫什么名字？") # AI会记得

进阶：持久化存储

import redis from langchain.memory import RedisChatMessageHistory class PersistentAgentSession: """支持持久化的Agent会话""" def __init__(self, session_id: str, redis_url: str = "redis://localhost:6379"): self.session_id = session_id # 使用Redis存储历史 self.history = RedisChatMessageHistory( session_id=session_id, url=redis_url ) self.memory = ConversationBufferWindowMemory( chat_memory=self.history, k=20, # 保留最近20轮 return_messages=True ) self.chain = ConversationChain( llm=llm, memory=self.memory )

记忆管理最佳实践：

• 限制历史长度（避免上下文过长）
• 为每个用户创建独立会话
• 定期清理过期会话
• 重要信息持久化存储
• 提供清空记忆的接口

陷阱5：工具调用死循环

❌ 错误现象

Agent: 我需要查询天气 → 调用Weather工具 Tool: 返回"北京今天晴" Agent: 我需要确认天气 → 再次调用Weather工具 Tool: 返回"北京今天晴" Agent: 我需要再次确认 → 又调用Weather工具 ...（无限循环）

🔍 问题分析

• Agent没有理解工具返回的结果
• Prompt没有明确的停止条件
• 缺乏最大迭代次数限制

✅ 正确做法

from langchain.agents import AgentExecutor, create_react_agent # 方法1：设置最大迭代次数 agent_executor = AgentExecutor( agent=agent, tools=tools, max_iterations=5, # ← 最多迭代5次 handle_parsing_errors=True, verbose=True ) # 方法2：在System Prompt中明确规则 system_prompt = """ 你是一个有用的助手。 【重要规则】 1. 如果工具已经返回了答案，直接使用它回答用户 2. 不要重复调用同一个工具超过2次 3. 如果无法获取信息，诚实地告诉用户 4. 最多进行3轮工具调用，然后必须给出最终答案 【工作流程】 思考 → 行动 → 观察 → 思考 → ... → 最终答案 """ # 方法3：自定义回调监控 from langchain.callbacks import BaseCallbackHandler class LoopDetectionHandler(BaseCallbackHandler): """检测并阻止死循环""" def __init__(self, max_same_tool_calls: int = 2): self.tool_call_count = {} self.max_same_tool_calls = max_same_tool_calls def on_tool_start(self, serialized, input_str, **kwargs): tool_name = serialized['name'] self.tool_call_count[tool_name] = self.tool_call_count.get(tool_name, 0) + 1 if self.tool_call_count[tool_name] > self.max_same_tool_calls: raise Exception( f"检测到死循环：工具 '{tool_name}' 已被调用 " f"{self.tool_call_count[tool_name]} 次" ) agent_executor = AgentExecutor( agent=agent, tools=tools, callbacks=[LoopDetectionHandler(max_same_tool_calls=2)] )

陷阱6：返回值格式错误

❌ 错误做法

def calculate(expression: str): """计算数学表达式""" result = eval(expression) return result # ← 返回数字类型 # Agent期望字符串，收到数字后解析失败

🔍 问题分析

• LangChain Agent期望工具返回字符串
• 返回其他类型会导致解析错误
• 格式不一致影响Agent理解

✅ 正确做法

def calculate(expression: str) -> str: """ 计算数学表达式 Args: expression: 数学表达式，如 '2 + 2' 或 '3 * (4 + 5)' Returns: 计算结果的字符串表示 """ try: # 安全计算（禁用危险函数） result = eval(expression, {"__builtins__": {}}, {}) return f"计算结果：{result}" # ← 始终返回字符串 except Exception as e: return f"计算错误：{str(e)}" # 更复杂的返回格式 def search_news(keyword: str) -> str: """搜索新闻""" articles = fetch_news(keyword) if not articles: return f"未找到关于'{keyword}'的新闻" # 格式化输出 output = f"找到 {len(articles)} 条相关新闻：\n\n" for i, article in enumerate(articles[:5], 1): output += f"{i}. {article['title']}\n" output += f" {article['summary']}\n" output += f" 来源：{article['source']} | " output += f"时间：{article['date']}\n\n" return output

返回值格式规范：

• 始终返回字符串
• 结构化输出（便于Agent解析）
• 包含必要的上下文信息
• 错误时返回清晰的错误消息
• 避免返回JSON（除非必要）

陷阱7：并发处理问题

❌ 错误做法

# 多个用户同时访问 @app.post("/chat") def chat(request: ChatRequest): session = get_session(request.user_id) return session.chat(request.question) # 问题：共享状态导致数据混乱

🔍 问题分析

• 全局变量被多线程修改
• 会话状态冲突
• 资源竞争
• 数据泄露风险

✅ 正确做法

from fastapi import FastAPI from threading import Lock app = FastAPI() # 线程安全的会话管理 class SessionManager: def __init__(self): self.sessions = {} self.lock = Lock() def get_session(self, user_id: str) -> AgentSession: with self.lock: if user_id not in self.sessions: self.sessions[user_id] = AgentSession(user_id) return self.sessions[user_id] def cleanup_expired(self): """清理过期会话""" with self.lock: # 实现清理逻辑 pass session_manager = SessionManager() @app.post("/chat") async def chat(request: ChatRequest): session = session_manager.get_session(request.user_id) response = await session.chat_async(request.question) return {"response": response} # 或使用异步锁 import asyncio class AsyncSessionManager: def __init__(self): self.sessions = {} self.lock = asyncio.Lock() async def get_session(self, user_id: str): async with self.lock: if user_id not in self.sessions: self.sessions[user_id] = AsyncAgentSession(user_id) return self.sessions[user_id]

并发安全要点：

• 使用锁保护共享状态
• 每个用户独立会话
• 异步环境使用asyncio.Lock
• 定期清理过期资源
• 考虑使用Redis等外部存储

陷阱8：安全性考虑不足

❌ 错误做法

# 直接执行用户输入 def execute_code(code: str): return eval(code) # ← 严重安全漏洞！ # 或未验证的工具参数 def send_email(to: str, content: str): send(to, content) # ← 可能被用于垃圾邮件

🔍 问题分析

• 代码注入攻击
• 敏感信息泄露
• 滥用工具造成损失
• 违反合规要求

✅ 正确做法

import re from typing import List # 1. 输入验证 def validate_email_address(email: str) -> bool: """验证邮箱地址格式""" pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return bool(re.match(pattern, email)) def send_email_tool(to: str, subject: str, content: str) -> str: """发送邮件（带安全检查）""" # 验证邮箱格式 if not validate_email_address(to): return "错误：无效的邮箱地址" # 检查内容长度 if len(content) > 5000: return "错误：邮件内容过长（最多5000字符）" # 过滤敏感词 sensitive_words = ["密码", "银行卡", "身份证"] for word in sensitive_words: if word in content: return "错误：邮件内容包含敏感信息" # 记录日志 log_email_sent(to, subject) # 发送邮件 try: actual_send_email(to, subject, content) return f"邮件已发送至 {to}" except Exception as e: return f"发送失败：{str(e)}" # 2. 权限控制 class ToolPermission: """工具权限管理""" def __init__(self): self.permissions = { "admin": ["all"], "user": ["search", "calculate", "query"], "guest": ["search"] } def check_permission(self, user_role: str, tool_name: str) -> bool: allowed = self.permissions.get(user_role, []) return tool_name in allowed or "all" in allowed # 3. 速率限制 from functools import wraps import time def rate_limit(max_calls: int, period: int = 60): """限制调用频率""" calls = [] def decorator(func): @wraps(func) def wrapper(*args, **kwargs): now = time.time() calls.append(now) # 移除过期的调用记录 while calls and calls[0] < now - period: calls.pop(0) if len(calls) > max_calls: return "错误：调用频率过高，请稍后重试" return func(*args, **kwargs) return wrapper return decorator @rate_limit(max_calls=10, period=60) # 每分钟最多10次 def search_tool(query: str): # ...

安全最佳实践：

• 永远不要直接执行用户输入
• 所有输入都要验证和清洗
• 实施权限控制和速率限制
• 记录所有敏感操作日志
• 定期安全审计

陷阱9：性能瓶颈

❌ 错误做法

# 每次都重新加载模型 def chat(question: str): llm = load_llm() # ← 耗时操作 return llm.invoke(question) # 或串行处理多个请求 for question in questions: result = agent.run(question) # ← 一个接一个

🔍 问题分析

• 重复初始化开销大
• 串行处理效率低
• 未使用缓存
• 响应时间长

✅ 正确做法

# 1. 单例模式（复用LLM实例） class LLMSingleton: _instance = None @classmethod def get_instance(cls): if cls._instance is None: cls._instance = ChatOpenAI(model="gpt-3.5-turbo") return cls._instance llm = LLMSingleton.get_instance() # 2. 缓存机制 from functools import lru_cache import hashlib @lru_cache(maxsize=1000) def cached_chat(question: str) -> str: """缓存常见问题""" return llm.invoke(question).content def smart_chat(question: str) -> str: """智能缓存""" # 生成问题指纹 fingerprint = hashlib.md5(question.encode()).hexdigest() # 检查缓存 if fingerprint in cache: return cache[fingerprint] # 调用LLM response = llm.invoke(question).content # 存入缓存 cache[fingerprint] = response return response # 3. 批量处理 import asyncio async def batch_process(questions: List[str]) -> List[str]: """并行处理多个问题""" tasks = [agent.ainvoke({"input": q}) for q in questions] results = await asyncio.gather(*tasks) return [r["output"] for r in results] # 4. 流式输出（提升用户体验） async def stream_chat(question: str): """流式返回结果""" async for chunk in llm.astream(question): yield chunk.content # FastAPI中使用 from fastapi.responses import StreamingResponse @app.post("/chat/stream") async def stream_chat_endpoint(request: ChatRequest): return StreamingResponse( stream_chat(request.question), media_type="text/plain" )

性能优化要点：

• 复用LLM实例（避免重复加载）
• 缓存常见问题
• 批量并行处理
• 流式输出改善体验
• 监控性能指标

陷阱10：测试覆盖不全

❌ 错误做法

# 只测试正常流程 def test_agent(): result = agent.run("你好") assert "你好" in result

🔍 问题分析

• 未测试边界情况
• 未测试错误处理
• 未测试并发场景
• 上线后才发现bug

✅ 正确做法

import pytest from unittest.mock import Mock, patch class TestAgent: """Agent全面测试""" def setup_method(self): """每个测试前的准备""" self.agent = create_test_agent() # 1. 正常功能测试 def test_basic_chat(self): """测试基本对话""" result = self.agent.run("你好") assert result is not None assert len(result) > 0 # 2. 边界情况测试 def test_empty_input(self): """测试空输入""" result = self.agent.run("") assert "请输入" in result or len(result) > 0 def test_very_long_input(self): """测试超长输入""" long_text = "测试" * 10000 result = self.agent.run(long_text) assert result is not None # 3. 错误处理测试 def test_tool_failure(self): """测试工具失败""" with patch('tools.search_web', side_effect=Exception("网络错误")): result = self.agent.run("搜索Python教程") assert "错误" in result or "失败" in result def test_timeout(self): """测试超时处理""" with patch('llm.invoke', side_effect=TimeoutError()): result = self.agent.run("测试") assert "超时" in result or "稍后" in result # 4. 并发测试 def test_concurrent_requests(self): """测试并发请求""" import concurrent.futures def chat_request(i): return self.agent.run(f"问题{i}") with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: futures = [executor.submit(chat_request, i) for i in range(20)] results = [f.result() for f in futures] assert len(results) == 20 assert all(r is not None for r in results) # 5. 集成测试 def test_full_workflow(self): """测试完整工作流""" # 模拟真实用户交互 conversation = [ "我想了解人工智能", "能详细介绍一下吗？", "有哪些应用场景？", "谢谢你的帮助" ] session = AgentSession("test_user") responses = [] for msg in conversation: response = session.chat(msg) responses.append(response) assert response is not None # 验证对话连贯性 assert len(responses) == len(conversation) # 6. 性能测试 def test_response_time(self): """测试响应时间""" import time start = time.time() result = self.agent.run("简单问题") elapsed = time.time() - start assert elapsed < 5.0 # 响应时间应小于5秒 print(f"响应时间：{elapsed:.2f}秒") # 运行测试 if __name__ == "__main__": pytest.main([__file__, "-v"])

测试覆盖清单：

• [ ] 正常功能测试
• [ ] 边界情况测试（空输入、超长输入）
• [ ] 错误处理测试（网络错误、超时）
• [ ] 并发测试（多用户同时访问）
• [ ] 集成测试（完整工作流）
• [ ] 性能测试（响应时间、吞吐量）
• [ ] 安全测试（注入攻击、权限绕过）

总结

🤔 互动环节

投票：你遇到过哪些陷阱？

• [ ] 工具调用死循环
• [ ] 记忆管理问题
• [ ] 并发安全问题
• [ ] 其他（留言说明）

留言：你还遇到过哪些Agent开发的坑？留言分享，帮助更多人避坑！

预告：下一篇《向量检索优化技巧》，敬请期待！

相关链接

• LangChain官方文档
• GitHub项目