1. 智能体基础概念解析
智能体（Agents）作为人工智能领域的重要概念，本质上是一个能够感知环境、自主决策并执行行动的计算实体。不同于传统的程序化系统，智能体的核心特征在于其自主性和适应性。根据不同的设计目标和应用场景，智能体可以呈现出多种形态和复杂度。
1.1 智能体的核心特性
智能体区别于普通软件系统的三大核心特性：
-
自主性（Autonomy）：智能体能够在没有直接人为干预的情况下运行，自主控制其内部状态和行为。例如，一个电商推荐系统智能体可以自动分析用户行为数据并生成个性化推荐，而不需要人工为每个用户手动配置规则。
-
反应性（Reactivity）：智能体能够感知环境变化并及时做出响应。以智能家居系统为例，当温度传感器检测到室温超过阈值时，空调控制智能体会立即启动制冷程序。
-
主动性（Proactiveness）：高级智能体不仅能对环境做出反应，还能主动追求目标。比如物流调度智能体不仅会响应运输需求，还会主动优化路线规划以降低成本。
1.2 智能体的分类体系
根据智能水平和功能特点，智能体可分为以下几个主要类型：
1.2.1 响应式智能体
最简单的智能体类型，采用"感知-动作"的直接映射模式。这类智能体没有内部状态记忆，每个决策仅基于当前感知的输入。典型的例子包括：
- 生产线上的物品分拣机器人
- 基本的自动化测试脚本
- 简单的规则引擎系统
1.2.2 基于模型的智能体
在响应式基础上增加了对环境的内部表示（世界模型），能够处理部分可观察的环境。例如：
- 棋牌类游戏AI（维护棋盘状态模型）
- 有限状态机实现的对话系统
- 基于环境模型的自动化测试框架
1.2.3 基于目标的智能体
引入显式的目标表示，能够规划行动序列以实现特定目标。典型应用包括：
- 路径规划导航系统
- 自动化任务调度系统
- 智能客服的问题解决流程
1.2.4 基于效用的智能体
在目标导向基础上增加了效用函数，能够评估不同行动方案的优劣。高级应用场景如：
- 金融投资决策系统
- 资源优化配置系统
- 复杂供应链管理系统
1.3 智能体的典型应用场景
现代智能体技术已广泛应用于各个领域：
-
对话系统：如客服聊天机器人、个人语音助手等。以ChatGPT为代表的LLM智能体能够处理复杂的自然语言交互。
-
推荐系统：电商平台、内容平台的个性化推荐引擎，能够学习用户偏好并主动推荐商品或内容。
-
自动化控制系统：包括工业自动化、智能家居、自动驾驶等领域的环境感知和决策系统。
-
游戏AI：从简单的NPC行为到复杂的战略决策系统，现代游戏大量运用各类智能体技术。
-
商业流程自动化：RPA（机器人流程自动化）工具实质上是特定领域的业务处理智能体。
2. 智能体架构设计与核心组件
构建一个完整的智能体系统需要精心设计其架构和各组件的协作方式。本节将深入解析智能体的核心架构模式及其实现要点。
2.1 智能体系统的基本架构
一个典型的智能体系统包含以下核心组件及其交互关系：
code复制[感知模块] → [决策引擎] → [执行模块]
↑ ↑ ↓
[环境接口] [知识库] [动作接口]
↑ ↓
[外部环境] ←------------------
2.1.1 感知模块
负责从环境获取信息并进行预处理，关键技术包括：
- 传感器数据采集（物理环境）
- API接口调用（数字环境）
- 自然语言理解（文本输入）
- 计算机视觉（图像/视频输入）
2.1.2 决策引擎
智能体的"大脑"，核心功能包括：
- 信息处理与推理
- 行动计划生成
- 冲突解决机制
- 学习与适应模块
2.1.3 执行模块
将决策转化为实际行动，涉及：
- 物理动作执行（机器人）
- API调用（软件系统）
- 自然语言生成（对话系统）
- 可视化输出（信息展示）
2.2 模型服务集成
现代智能体通常需要集成各类AI模型服务，常见的集成方式包括：
2.2.1 本地模型部署
使用Ollama等工具本地运行模型的优势：
- 数据隐私性好
- 响应延迟低
- 可完全控制模型行为
典型部署步骤：
- 下载并安装Ollama
- 拉取所需模型（如llama2、mistral等）
- 通过REST API与模型交互
bash复制# Ollama基本操作命令
ollama pull llama2 # 下载模型
ollama list # 查看已安装模型
ollama run llama2 # 运行模型交互
2.2.2 云API服务
使用OpenAI等云服务的考虑因素：
- 模型能力强大
- 无需本地计算资源
- 但存在数据隐私和成本问题
2.3 工具系统设计
工具（Tools）是智能体能力的扩展机制，良好的工具系统设计应遵循以下原则：
- 单一职责原则：每个工具应只完成一个明确的任务
- 接口标准化：统一的输入输出格式便于智能体调用
- 完备的文档：清晰的工具描述有助于智能体正确选择工具
- 错误处理机制：工具应能处理异常输入并提供有意义的错误信息
2.3.1 工具实现示例
以计算器工具为例，一个健壮的实现应包含：
python复制def advanced_calculator(input_data):
"""
支持多种数学运算的高级计算器工具
参数:
input_data (str/dict): 包含运算要求的JSON字符串或字典
格式示例:
'{"expression": "(5+3)*2", "precision": 2}'
或
{"expression": "sqrt(16)", "precision": 0}
返回:
dict: 包含结果和状态信息的字典
示例: {"result": 16.0, "status": "success", "error": None}
"""
try:
if isinstance(input_data, str):
params = json.loads(input_data)
else:
params = input_data
# 安全评估数学表达式
allowed_chars = set('0123456789+-*/.() sqrt')
if not all(c in allowed_chars for c in params['expression']):
return {"status": "error", "error": "非法字符"}
# 计算表达式
result = eval(params['expression'], {'__builtins__': None}, {})
# 处理精度
precision = params.get('precision', 2)
rounded_result = round(result, precision) if precision else int(result)
return {
"result": rounded_result,
"status": "success",
"error": None
}
except Exception as e:
return {
"result": None,
"status": "error",
"error": str(e)
}
2.3.2 工具注册与管理
工具系统的核心是ToolBox类，它负责工具的注册和描述管理：
python复制class ToolBox:
def __init__(self):
self.tools = {}
self.descriptions = {}
def register(self, tool_func, description=None):
"""注册工具并存储其描述"""
tool_name = tool_func.__name__
self.tools[tool_name] = tool_func
self.descriptions[tool_name] = description or tool_func.__doc__
def get_tool(self, name):
"""获取指定工具"""
return self.tools.get(name)
def list_tools(self):
"""获取所有工具描述"""
return "\n".join(
f"{name}: {desc}"
for name, desc in self.descriptions.items()
)
3. 智能体核心实现详解
本节将深入讲解智能体的具体实现过程，从环境准备到完整系统搭建的全流程。
3.1 开发环境配置
3.1.1 Python环境设置
推荐使用conda或venv创建隔离的Python环境：
bash复制# 使用conda创建环境
conda create -n ai_agent python=3.10
conda activate ai_agent
# 或使用venv
python -m venv ai_agent_env
source ai_agent_env/bin/activate # Linux/Mac
ai_agent_env\Scripts\activate # Windows
安装基础依赖包：
bash复制pip install requests python-dotenv termcolor
3.1.2 Ollama本地部署
- 从Ollama官网下载对应平台的安装包
- 安装后验证服务是否正常运行：
bash复制ollama --version
curl http://localhost:11434/api/tags # 查看可用模型
- 下载所需模型（以llama2为例）：
bash复制ollama pull llama2
- 测试模型API：
bash复制curl -X POST http://localhost:11434/api/generate -d '{
"model": "llama2",
"prompt": "介绍一下人工智能"
}'
3.2 模型服务封装
对Ollama API的封装应考虑以下方面：
- 连接稳定性：重试机制和超时设置
- 错误处理：全面的异常捕获
- 性能监控：记录响应时间和成功率
- 格式转换：统一输入输出格式
改进后的模型服务实现：
python复制import requests
import json
import time
from typing import Optional, Dict, Any
class OllamaService:
def __init__(self, base_url: str = "http://localhost:11434",
timeout: int = 30, max_retries: int = 3):
self.base_url = base_url.rstrip('/')
self.timeout = timeout
self.max_retries = max_retries
self.session = requests.Session()
def generate(self, model: str, prompt: str,
system: Optional[str] = None,
format: str = "json",
temperature: float = 0.7,
**kwargs) -> Dict[str, Any]:
"""
生成文本的增强方法
参数:
model: 模型名称
prompt: 用户提示
system: 系统提示
format: 返回格式(json/text)
temperature: 创造性控制
**kwargs: 其他API参数
返回:
包含响应和元数据的字典
"""
url = f"{self.base_url}/api/generate"
payload = {
"model": model,
"prompt": prompt,
"system": system,
"format": format,
"options": {
"temperature": temperature,
**kwargs
}
}
headers = {"Content-Type": "application/json"}
for attempt in range(self.max_retries):
try:
start_time = time.time()
response = self.session.post(
url,
json=payload,
headers=headers,
timeout=self.timeout
)
response.raise_for_status()
# 处理流式响应
full_response = ""
for line in response.iter_lines():
if line:
chunk = json.loads(line)
full_response += chunk.get("response", "")
return {
"response": full_response,
"latency": time.time() - start_time,
"status": "success"
}
except requests.exceptions.RequestException as e:
if attempt == self.max_retries - 1:
return {
"response": None,
"error": str(e),
"status": "error"
}
time.sleep(1 * (attempt + 1))
3.3 智能体决策逻辑实现
智能体的核心决策流程包括：
- 意图识别：分析用户输入的潜在意图
- 工具选择：根据意图匹配合适的工具
- 输入解析：提取工具所需的参数
- 结果整合：将工具输出转化为用户友好的响应
3.3.1 系统提示设计
精心设计的系统提示对智能体性能至关重要。一个好的提示应包含：
- 角色定义：明确智能体的身份和能力范围
- 工具规范：详细描述每个工具的用途和使用方式
- 输出格式：严格规定响应数据结构
- 错误处理：指导如何处理无法理解或执行的请求
改进后的系统提示模板：
python复制AGENT_SYSTEM_PROMPT = """你是一个专业AI助手，可以访问特定工具完成任务。请严格遵守以下规则：
# 身份与能力
- 身份：专业AI助手
- 能力范围：{capabilities}
- 限制：{limitations}
# 响应格式
必须使用严格JSON格式：
{{
"thoughts": "你的思考过程",
"tool": "工具名称|none",
"input": "工具输入|自然语言响应",
"confidence": 0-1的置信度
}}
# 可用工具
{tools_description}
# 处理流程
1. 分析用户请求的意图
2. 评估是否适合使用工具处理
3. 如使用工具，提取精确参数
4. 如不使用工具，生成友好自然语言响应
# 特别注意
- 数学计算必须使用calculator工具
- 字符串操作必须使用string_tools
- 不确定时返回低置信度
- 超出能力范围时明确告知用户
当前时间：{current_time}
"""
3.3.2 智能体类实现
增强版的智能体实现包含以下改进：
- 对话历史：维护上下文记忆
- 置信度评估：量化决策可靠性
- 输入验证：确保工具参数有效性
- 性能统计：记录决策时间和成功率
python复制from datetime import datetime
import hashlib
class EnhancedAgent:
def __init__(self, model_service, tools: list, model_name: str):
self.model = model_service
self.model_name = model_name
self.toolbox = ToolBox()
for tool in tools:
self.toolbox.register(tool)
self.conversation_history = []
self.stats = {
"total_requests": 0,
"tool_usage": {},
"avg_response_time": 0
}
def generate_session_id(self):
"""生成唯一会话ID"""
return hashlib.md5(str(datetime.now()).encode()).hexdigest()[:8]
def think(self, user_input: str) -> dict:
"""增强的决策方法"""
start_time = time.time()
session_id = self.generate_session_id()
# 准备系统提示
tools_desc = self.toolbox.list_tools()
system_prompt = AGENT_SYSTEM_PROMPT.format(
capabilities="数学计算、字符串处理、信息查询",
limitations="不能访问外部网络或私人数据",
tools_description=tools_desc,
current_time=str(datetime.now())
)
# 构建对话上下文
context = "\n".join(
f"User: {msg['input']}\nAI: {msg['response']}"
for msg in self.conversation_history[-3:]
)
full_prompt = f"Context:\n{context}\n\nNew Input:\n{user_input}"
# 调用模型
response = self.model.generate(
model=self.model_name,
prompt=full_prompt,
system=system_prompt,
temperature=0.3
)
# 解析响应
try:
decision = json.loads(response["response"])
if not all(k in decision for k in ["tool", "input"]):
raise ValueError("Invalid response format")
decision["session_id"] = session_id
decision["processing_time"] = time.time() - start_time
# 更新统计
self._update_stats(decision.get("tool"))
return decision
except Exception as e:
return {
"tool": "none",
"input": "抱歉，我遇到了处理问题，请稍后再试。",
"confidence": 0,
"error": str(e)
}
def _update_stats(self, tool_name: str):
"""更新性能统计"""
self.stats["total_requests"] += 1
if tool_name and tool_name != "none":
self.stats["tool_usage"][tool_name] = \
self.stats["tool_usage"].get(tool_name, 0) + 1
def execute(self, decision: dict) -> str:
"""执行工具并返回结果"""
tool_name = decision["tool"]
if tool_name == "none":
return decision["input"]
tool = self.toolbox.get_tool(tool_name)
if not tool:
return f"错误：找不到工具 {tool_name}"
try:
# 预处理工具输入
if isinstance(decision["input"], str):
tool_input = json.loads(decision["input"])
else:
tool_input = decision["input"]
# 执行工具
result = tool(tool_input)
# 记录对话
self.conversation_history.append({
"input": decision.get("original_input", ""),
"response": str(result),
"timestamp": str(datetime.now())
})
return result
except Exception as e:
return f"执行工具时出错: {str(e)}"
4. 智能体系统优化与进阶话题
构建基础智能体只是第一步，要让智能体在实际应用中表现优异，还需要考虑诸多优化和扩展方向。
4.1 性能优化策略
4.1.1 响应速度优化
- 模型量化：使用4-bit或8-bit量化减小模型大小
- 缓存机制：缓存常见请求的响应
- 预加载：提前加载常用工具和资源
- 并行处理：对独立任务采用多线程/进程
python复制from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
class OptimizedAgent(EnhancedAgent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.executor = ThreadPoolExecutor(max_workers=4)
@lru_cache(maxsize=100)
def cached_think(self, user_input: str) -> dict:
"""带缓存的思考方法"""
return self.think(user_input)
async def async_execute(self, decision: dict):
"""异步执行工具"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
self.executor,
self.execute,
decision
)
4.1.2 内存管理
- 对话历史修剪：只保留最近N条对话
- 工具懒加载：按需加载工具而非启动时全部加载
- 资源清理：定期清理临时文件和缓存
4.2 可靠性增强
4.2.1 错误处理机制
- 输入验证层：在调用工具前验证参数
- 重试策略：对临时性错误自动重试
- 降级方案：主工具失败时使用备用方案
- 超时控制：防止长时间无响应
python复制from tenacity import retry, stop_after_attempt, wait_exponential
class RobustAgent(EnhancedAgent):
@retry(stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10))
def reliable_execute(self, decision: dict) -> str:
"""带重试机制的工具执行"""
try:
with timeout(5): # 5秒超时
return self.execute(decision)
except TimeoutError:
return "操作超时，请稍后再试"
except Exception as e:
self.log_error(e)
raise
4.2.2 监控与日志
- 性能指标收集：响应时间、成功率等
- 异常警报：错误率超过阈值时通知
- 审计日志：记录所有重要操作
- 使用分析：统计工具使用频率和模式
python复制import logging
from prometheus_client import Counter, Histogram
class MonitoredAgent(EnhancedAgent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.logger = logging.getLogger("AgentLogger")
self.request_counter = Counter("agent_requests", "Total requests")
self.error_counter = Counter("agent_errors", "Total errors")
self.latency_histogram = Histogram(
"agent_latency",
"Response latency",
buckets=[0.1, 0.5, 1, 2, 5]
)
def think(self, user_input: str) -> dict:
self.request_counter.inc()
start_time = time.time()
try:
result = super().think(user_input)
latency = time.time() - start_time
self.latency_histogram.observe(latency)
if result.get("error"):
self.error_counter.inc()
self.logger.error(
f"Request failed: {result['error']}",
extra={"input": user_input}
)
return result
except Exception as e:
self.error_counter.inc()
self.logger.exception("Unexpected error")
raise
4.3 进阶功能扩展
4.3.1 多智能体协作
- 角色分工：不同智能体负责不同专业领域
- 通信协议：定义智能体间的交互方式
- 冲突解决：处理智能体间的意见分歧
- 协调机制：统筹多个智能体的行动
python复制class MultiAgentSystem:
def __init__(self, agents: dict):
"""
参数:
agents: {"角色": 智能体实例} 的字典
"""
self.agents = agents
self.coordinator = EnhancedAgent(...)
def solve_complex_task(self, task_description: str) -> dict:
# 协调器分析任务并分配子任务
plan = self.coordinator.think(
f"任务：{task_description}\n"
f"可用专家：{', '.join(self.agents.keys())}\n"
"请制定执行计划"
)
results = {}
for subtask, agent_role in plan["subtasks"].items():
agent = self.agents[agent_role]
results[subtask] = agent.execute(subtask)
# 综合各专家结果
final_result = self.coordinator.think(
f"原始任务：{task_description}\n"
f"专家结果：{json.dumps(results)}\n"
"请整合最终解决方案"
)
return {
"subtask_results": results,
"final_solution": final_result
}
4.3.2 持续学习机制
- 反馈循环：收集用户反馈改进表现
- 自动微调：基于交互数据调整模型参数
- 知识更新：定期刷新内置知识库
- 行为分析：识别并修正不良模式
python复制class LearningAgent(EnhancedAgent):
def __init__(self, *args, feedback_db=None, **kwargs):
super().__init__(*args, **kwargs)
self.feedback_db = feedback_db or []
self.learning_rate = 0.1
def receive_feedback(self, session_id: str, rating: int, comments: str = ""):
"""接收用户反馈并存储"""
self.feedback_db.append({
"session_id": session_id,
"rating": rating,
"comments": comments,
"timestamp": datetime.now()
})
def adjust_behavior(self):
"""根据反馈调整行为"""
if not self.feedback_db:
return
recent_feedback = [
fb for fb in self.feedback_db
if datetime.now() - fb["timestamp"] < timedelta(days=7)
]
if not recent_feedback:
return
avg_rating = sum(fb["rating"] for fb in recent_feedback) / len(recent_feedback)
# 根据评分调整温度参数
if avg_rating < 3:
self.model.temperature = max(
0.1,
self.model.temperature - self.learning_rate
)
elif avg_rating > 4:
self.model.temperature = min(
1.0,
self.model.temperature + self.learning_rate
)
# 分析负面反馈中的常见问题
common_issues = Counter(
fb["comments"] for fb in recent_feedback
if fb["rating"] < 3 and fb["comments"]
)
if common_issues:
self.update_guidelines(common_issues)
def update_guidelines(self, issues: dict):
"""更新系统提示中的指导原则"""
new_rules = "\n".join(
f"- 避免：{issue}"
for issue, count in issues.most_common(3)
)
current_prompt = self.model.system_prompt
if "特别注意" in current_prompt:
updated_prompt = current_prompt.replace(
"特别注意",
f"特别注意\n{new_rules}\n"
)
self.model.update_system_prompt(updated_prompt)
5. 智能体开发实战经验与避坑指南
在实际开发智能体系统的过程中，会遇到各种预料之外的挑战和陷阱。本节将分享从实战中积累的重要经验和常见问题的解决方案。
5.1 工具设计最佳实践
5.1.1 工具接口设计原则
- 强类型检查：在工具入口处验证输入类型
- 参数默认值：为可选参数提供合理的默认值
- 错误代码：定义清晰的错误代码体系
- 结果标准化：统一成功/失败的结果格式
python复制from pydantic import BaseModel, validator
from enum import Enum
class ToolErrorCode(Enum):
INVALID_INPUT = 1001
OPERATION_FAILED = 1002
RESOURCE_UNAVAILABLE = 1003
class ToolResult(BaseModel):
success: bool
data: Optional[Any]
error_code: Optional[ToolErrorCode]
message: Optional[str]
class CalculatorInput(BaseModel):
expression: str
precision: int = 2
@validator('expression')
def validate_expression(cls, v):
allowed_chars = set('0123456789+-*/.() sqrt')
if not all(c in allowed_chars for c in v):
raise ValueError("表达式包含非法字符")
return v
def safe_calculator(input_data: Union[str, dict]) -> ToolResult:
try:
# 输入验证和转换
if isinstance(input_data, str):
input_dict = json.loads(input_data)
else:
input_dict = input_data
validated_input = CalculatorInput(**input_dict)
# 执行计算
result = eval(validated_input.expression, {'__builtins__': None}, {})
rounded = round(result, validated_input.precision)
return ToolResult(
success=True,
data={"result": rounded},
error_code=None,
message="计算成功"
)
except ValueError as e:
return ToolResult(
success=False,
data=None,
error_code=ToolErrorCode.INVALID_INPUT,
message=str(e)
)
except Exception as e:
return ToolResult(
success=False,
data=None,
error_code=ToolErrorCode.OPERATION_FAILED,
message=f"计算错误: {str(e)}"
)
5.1.2 工具测试策略
- 单元测试：验证每个工具在各种输入下的行为
- 集成测试：测试工具与智能体的集成效果
- 模糊测试：使用随机输入测试鲁棒性
- 性能测试：评估工具的执行效率
python复制import pytest
from hypothesis import given, strategies as st
class TestCalculator:
@given(
st.one_of(
st.just('1+1'),
st.just('3*5-2'),
st.just('sqrt(16)')
),
st.integers(min_value=0, max_value=5)
)
def test_valid_inputs(self, expr, precision):
input_data = {"expression": expr, "precision": precision}
result = safe_calculator(input_data)
assert result.success is True
assert isinstance(result.data["result"], (int, float))
def test_invalid_chars(self):
result = safe_calculator({"expression": "import os", "precision": 2})
assert result.success is False
assert result.error_code == ToolErrorCode.INVALID_INPUT
def test_division_by_zero(self):
result = safe_calculator({"expression": "1/0", "precision": 2})
assert result.success is False
assert result.error_code == ToolErrorCode.OPERATION_FAILED
5.2 模型交互常见问题
5.2.1 输出格式不一致
问题：模型可能返回不符合指定格式的响应
解决方案：
- 在系统提示中强调格式要求
- 实现格式验证和自动修正层
- 对常见格式错误建立修正规则库
python复制def normalize_response(raw_response: str) -> dict:
"""标准化模型响应格式"""
# 尝试直接解析
try:
return json.loads(raw_response)
except json.JSONDecodeError:
pass
# 常见格式问题的修正
fixes = [
(r"(\w+)\s*:\s*([^'\n]+?)(?=\s*\w+\s*:|$)", r'"\1": "\2"'), # 无引号的key: value
(r"'(.*?)'", r'"\1"'), # 单引号转双引号
(r"True|False", lambda m: m.group().lower()), # Python布尔转JSON布尔
(r"None", "null"), # Python None转JSON null
]
for pattern, replacement in fixes:
raw_response = re.sub(pattern, replacement, raw_response)
try:
return json.loads(raw_response)
except json.JSONDecodeError as e:
return {
"tool": "none",
"input": f"无法解析响应: {str(e)}",
"confidence": 0
}
5.2.2 工具选择偏差
问题：模型可能过度使用或忽视某些工具
解决方案：
- 在系统提示中明确工具选择标准
- 实现工具使用统计和自动平衡
- 引入工具选择置信度阈值
python复制class BalancedAgent(EnhancedAgent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.tool_weights = {name: 1.0 for name in self.toolbox.tools}
def think(self, user_input: str) -> dict:
decision = super().think(user_input)
# 调整工具选择置信度
if decision["tool"] != "none":
base_confidence = decision.get("confidence", 1)
weight = self.tool_weights[decision["tool"]]
adjusted_confidence = base_confidence * weight
if adjusted_confidence < 0.5: # 低于阈值则不使用工具
decision.update({
"tool": "none",
"input": "我不确定如何处理这个请求",
"confidence": adjusted_confidence
})
else:
decision["confidence"] = adjusted_confidence
return decision
def update_tool_weights(self):
"""根据工具使用效果动态调整权重"""
total_usage = sum(self.stats["tool_usage"].values())
if not total_usage:
return
avg_success = self.stats["success_rate"]
for tool, count in self.stats["tool_usage"].items():
usage_rate = count / total_usage
success_rate = self.stats["tool_success"].get(tool, 0) / max(1, count)
# 计算新权重：使用率与成功率平衡
new_weight = 0.7 * success_rate + 0.3 * (1 - usage_rate)
self.tool_weights[tool] = max(0.1, min(1.0, new_weight))
5.3 生产环境部署考量
5.3.1 安全防护措施
- 输入消毒：防止注入攻击
- 访问控制：API密钥和权限管理
- 数据脱敏：敏感信息过滤
- 请求限流：防止滥用
python复制from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import APIKeyHeader
from slowapi import Limiter
from slowapi.util import get_remote_address
app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
api_key_header = APIKeyHeader(name="X-API-KEY")
def sanitize_input(text: str) -> str:
"""移除潜在危险字符"""
return re.sub(r"[;\\'\"]", "", text[:1000])
async def validate_api_key(api_key: str = Depends(api_key_header)):
if api_key != os.getenv("API_KEY"):
raise HTTPException(status_code=403, detail="无效API密钥")
@app.post("/chat")
@limiter.limit("10/minute")
async def chat_endpoint(
request: Request,
query: str,
_=Depends(validate_api_key)
):
try:
safe_query = sanitize_input(query)
response = agent.work(safe_query)
return {"response": response}
except Exception as e:
raise HTTPException(500, "处理请求时出错")
5.3.2 性能与扩展性
- 水平扩展：无状态设计支持多实例
- 资源隔离：关键工具独立部署
- 异步处理：长时间操作异步执行
- 自动伸缩：根据负载调整资源
python复制import redis
from celery import Celery
# 消息队列配置
celery_app = Celery(
'agent_tasks',
broker=os.getenv("REDIS_URL"),
backend=os.getenv("REDIS_URL")
)
# Redis缓存
cache = redis.Redis.from_url(os.getenv("REDIS_URL"))
@celery_app.task(bind=True)
def process_async_request(self, query: str):
"""异步处理复杂请求"""
try:
result = agent.work(query)
cache.set(self.request.id, json.dumps(result), ex=3600)
return result
except Exception as e:
cache.set(self.request.id, json.dumps({"error": str(e)}), ex=3600)
raise
@app.post("/async-chat")
async def async_chat_endpoint(query: str):
task = process_async_request.delay(query)
return {"task_id": task.id}
@app.get("/result/{task_id}")
async def get_result(task_id: str):
result = cache.get(task_id)
if not result:
raise HTTPException(404, "结果不存在或已过期")
return json.loads(result)