最佳实践
更新: 4/9/2026 字数: 0 字 时长: 0 分钟
本章总结 LangChain Agent 开发的最佳实践，涵盖性能、安全、可维护性等方面。
Prompt 工程
Prompt 模板结构
python
# 推荐的 Prompt 模板结构
SYSTEM_PROMPT = """# 角色定义
你是一个[角色名称]，专门负责[主要职责]。
# 能力范围
你可以：
1. [能力1]
2. [能力2]
3. [能力3]
# 工作原则
- [原则1]
- [原则2]
# 输出格式
请按以下格式输出：[输出格式说明]
# 限制
- 不要[禁止行为]
- 遇到[边界情况]时，请[处理方式]
"""Few-shot 示例
python
SYSTEM_PROMPT = """你是一个翻译助手，将中文翻译成英文。
示例：
输入：你好世界
输出：Hello World
输入：今天天气真好
输出：The weather is nice today
输入：{input}
输出："""工具设计
工具设计检查清单
工具设计规范
═══════════════════════════════════════════════════════════════
命名规范
────────────────────────────────────────────────────────────
✅ 使用 snake_case
✅ 动词或动词短语开头：get_, search_, create_
✅ 名词描述对象：get_user_info, search_products
参数规范
────────────────────────────────────────────────────────────
✅ 类型注解
✅ 默认值（可选参数）
✅ Pydantic 验证
✅ 完整描述
返回规范
────────────────────────────────────────────────────────────
✅ 返回字符串
✅ 结构化输出
✅ 包含状态信息
✅ 友好错误提示
═══════════════════════════════════════════════════════════════工具分组
python
# 按功能分组工具
TOOLS = {
"search": [search_web, search_news, search_academic],
"communication": [send_email, send_sms, send_notification],
"data": [query_database, get_analytics, generate_report],
}
def get_tools_by_permission(user_role: str) -> list:
permissions = {
"admin": ["*"],
"user": ["search", "data"],
"guest": ["search"],
}
return permissions.get(user_role, [])错误处理
分层错误处理
python
@tool
def robust_tool(input_data: str) -> str:
"""带错误处理的工具"""
try:
result = call_api(input_data)
return result
except ValidationError as e:
return f"参数错误：{str(e)}"
except RateLimitError:
return f"请求过于频繁，请稍后再试"
except TimeoutError:
return f"请求超时，请稍后重试"
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
return f"发生未知错误"重试装饰器
python
import time
from functools import wraps
def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except (RateLimitError, TimeoutError) as e:
if attempt < max_attempts - 1:
wait = delay * (backoff ** attempt)
time.sleep(wait)
return wrapper
return decorator性能优化
缓存策略
python
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_weather(city: str) -> str:
"""带缓存的天气查询"""
return fetch_weather_from_api(city)消息截断
python
from langchain_core.messages import trim_messages, get_buffer_string
def trim_history(messages: list, max_tokens: int = 8000) -> list:
"""截断消息历史"""
buffer = get_buffer_string(messages)
if len(buffer) > max_tokens * 4:
trimmer = trim_messages(
max_tokens=max_tokens,
strategy="last",
include_system=True
)
return trimmer.invoke(messages)
return messages安全考虑
输入验证
python
from pydantic import BaseModel, Field
class SafeInput(BaseModel):
query: str = Field(..., max_length=500)
email: str | None = None
def sanitize_input(user_input: str) -> str:
"""清理用户输入"""
dangerous = ["<script>", "DROP TABLE", "rm -rf"]
for pattern in dangerous:
user_input = user_input.replace(pattern, "")
return user_input.strip()敏感信息处理
python
import re
def mask_sensitive(text: str) -> str:
"""脱敏处理"""
text = re.sub(r'(\d{3})\d{4}(\d{4})', r'\1****\2', text) # 手机号
text = re.sub(r'([a-zA-Z0-9._%+-]+)@([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})', r'\1***@\2', text)
return text测试策略
单元测试
python
# tests/test_tools.py
import pytest
from tools import get_weather, calculator
class TestWeatherTool:
def test_get_weather_known_city(self):
result = get_weather("北京")
assert "北京" in result
assert "°C" in result
def test_get_weather_unknown_city(self):
result = get_weather("不存在城市")
assert "不支持" in result or "未知" in result
class TestCalculator:
def test_addition(self):
assert calculator("2 + 3") == "5"
def test_invalid_expression(self):
assert "错误" in calculator("2 +")集成测试
python
# tests/test_agent.py
import pytest
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
@pytest.fixture
def agent():
return create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather, calculator],
system_prompt="你是一个助手",
checkpointer=InMemorySaver()
)
class TestAgent:
def test_simple_query(self, agent):
result = agent.invoke({
"messages": [{"role": "user", "content": "你好"}]
})
assert len(result["messages"]) >= 2
def test_memory(self, agent):
config = {"configurable": {"thread_id": "test_memory"}}
agent.invoke(
{"messages": [{"role": "user", "content": "记住我喜欢蓝色"}]},
config=config
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "我喜欢什么颜色"}]},
config=config
)
assert "蓝色" in result["messages"][-1]["content"]本章小结
最佳实践要点
────────────────────────────────────────────────────────
Prompt 工程
✅ 清晰的角色定义
✅ 详细的工具描述
✅ 适当的示例
工具设计
✅ 命名规范
✅ 类型注解
✅ 参数验证
错误处理
✅ 分层处理
✅ 重试机制
✅ 友好提示
性能优化
✅ 缓存策略
✅ 消息截断
✅ 异步执行
安全
✅ 输入验证
✅ 敏感信息脱敏
✅ 权限控制