背景简介
Agentic AI 系列课程已经走过了五个阶段，我们从基础的 ReAct 范式实现，到多 Agent 协同，再到模块化架构，逐步优化了 Router Agent 和 Executor Agent。今天，我们将聚焦于 Validator Agent 的优化，这是确保整个 Agent 系统准确性和可靠性的关键环节。
Validator Agent 负责验证工具调用的参数和结果是否符合用户意图，是整个 Agent 系统的“质量检查员”。通过优化 Validator Agent，我们可以显著提升系统的准确性，减少错误传播，提高用户体验。
前置信息
- 完成 【Agentic AI - 第一课 - 从零实现基于 ReAct 范式的智能 Agent】
- 完成 【Agentic AI - 第二课 - 多Agent协同】
- 完成 【Agentic AI - 第三课 - 多Agent协同模块化】
- 完成 【Agentic AI - 第四课 - 优化 Router Agent 实现高效意图识别】
- 完成 【Agentic AI - 第五课 - 优化 Executor Agent】
详细信息
Validator Agent 的核心作用
Validator Agent 在整个 Agent 系统中扮演着“质量检查员”的角色。它的主要职责是验证工具调用的参数和结果是否符合用户意图，确保系统的准确性和可靠性。
主要功能
- 参数验证：检查传递给工具的参数是否正确
- 结果验证：验证工具执行结果是否符合预期
- 错误诊断：识别并分析错误原因
- 修正建议：提供具体的修正建议
- 性能监控：跟踪验证过程的性能指标
优化后的 Validator Agent 架构
- 协议化设计： 优化后的 Validator Agent 采用了协议化设计，通过 LLMClientProtocol 定义了与大语言模型交互的标准接口。这种设计确保了不同 LLM 实现的一致性，提高了代码的可维护性和扩展性。
- 配置管理： 通过 ValidatorConfig 类集中管理所有配置参数，包括模型参数、超时设置和系统提示等，支持灵活的参数调整。配置类包含了模型名称、温度参数、超时时间和最大重试次数等关键设置。
- 性能指标监控： ValidatorMetrics 类提供了全面的性能监控功能，通过这些指标，我们可以实时监控验证过程的性能表现，包括总验证次数、通过率、失败原因分布和平均延迟等。
核心验证逻辑
验证流程
- 构建提示词：根据用户输入、工具参数和执行结果构建验证提示词
- 调用 LLM：通过协议接口调用大语言模型进行验证
- 解析结果：解析 LLM 返回的验证结果
- 重试机制：支持自动重试，提高验证成功率
- 指标统计：记录验证过程的各项指标
代码示例
优化后 Validator Agent 代码
# app/ai_agent/agent_validator.py
"""验证Agent模块
提供基于大语言模型的工具调用验证功能，用于审核参数和结果是否符合用户意图。
主要功能包括结果验证、错误诊断、修正建议、性能监控和异步处理。
"""
import logging
import asyncio
from typing import Protocol, List, Dict, Any, Optional, runtime_checkable, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
from time import perf_counter
logger = logging.getLogger(__name__)
@runtime_checkable
class LLMClientProtocol(Protocol):
"""大语言模型客户端协议定义
定义了与大语言模型交互的标准接口，确保不同LLM实现的一致性
"""
async def chat(self, messages: List[Dict[str, str]], tools: List[Dict]) -> Any:
"""与LLM进行对话交互
Args:
messages: 对话消息列表，包含角色和内容
tools: 可用工具列表
Returns:
LLM的响应结果，格式取决于具体实现
"""
...
@dataclass
class ValidatorConfig:
"""验证Agent配置类
包含LLM参数、超时设置和系统提示等配置，用于控制验证Agent的行为
Attributes:
model: 使用的模型名称
temperature: 温度参数，控制输出的随机性
timeout: 超时时间（秒）
max_retries: 最大重试次数
validator_prompt_template: 验证提示词模板
"""
model: str = "llama3.1:8b"
temperature: float = 0.2
timeout: float = 30.0
max_retries: int = 3
validator_prompt_template: str = (
"你是参数审核员。请判断工具调用是否成功且符合用户意图。\n"
"你需要查看：【用户问题】、【工具参数】、【工具执行结果】。\n\n"
"重要规则：\n"
"1. 【中文城市名处理】：如果工具参数中的城市名与用户问题不一致，"
"必须指出错误并提供正确的城市名（逐字复制用户问题中的城市名）。\n"
"2. 【表达式处理】：计算器工具的表达式必须只包含数字和基本运算符，"
"不能包含 π、e 等特殊字符，应转换为数值（如 3.14159）。\n\n"
"输出要求：\n"
"1. 成功：只输出 'PASS'。\n"
"2. 失败：输出 'FAIL: 原因和建议修正方案'。\n\n"
"【示例】\n"
"示例 1 (正确提取):\n"
"用户: 兰州的天气\n"
"参数: {{'city': '兰州'}}\n"
"结果: 晴，25度\n"
"结果: PASS\n\n"
"示例 2 (提取错误导致未找到城市):\n"
"用户: 兰州的天气\n"
"参数: {{'city': '勗安'}}\n"
"结果: 未找到城市: 勗安\n"
"结果: FAIL: 工具未找到城市 '勗安'。用户问的是 '兰州'，请直接使用 '兰州' 作为参数。\n\n"
"示例 3 (中文乱码/错字):\n"
"用户: 上海天气\n"
"参数: {{'city': '上晦'}}\n"
"结果: 无法获取天气信息\n"
"结果: FAIL: 参数城市名 '上晦' 错误，请修正为 '上海'。\n\n"
"示例 4 (错误的表达式):\n"
"用户: 计算一个半径为3的圆的面积\n"
"参数: {{'expression': 'π*3^2'}}\n"
"结果: 表达式无效或包含非法操作:\n"
"结果: FAIL: 表达式非法，转化为纯数字的公式，不使用特殊字符，例如'3.14159 * 3 * 3'。\n\n"
"【当前任务】\n"
"工具类型: {tool_type}\n"
"工具名称: {tool_name}\n"
"用户: {user_message}\n"
"参数: {tool_args}\n"
"工具结果: {tool_result}\n"
"结果:"
)
def __post_init__(self):
"""初始化后验证配置参数的有效性
Raises:
ValueError: 当配置参数不符合要求时抛出
"""
self._validate()
def _validate(self):
"""验证配置参数的有效性
Raises:
ValueError: 当temperature不在0-2范围内或timeout不大于0时
"""
if not (0 <= self.temperature <= 2):
raise ValueError("temperature 必须在 0-2 之间")
if self.timeout <= 0:
raise ValueError("timeout 必须大于 0")
if self.max_retries < 0:
raise ValueError("max_retries 不能为负数")
@dataclass
class ValidatorMetrics:
"""验证指标统计类
用于跟踪和统计验证Agent的性能指标，包括验证次数、通过率、失败原因分布等
Attributes:
total_validations: 总验证次数
passed_validations: 通过验证次数
failed_validations: 失败验证次数
timeout_count: 超时次数
failure_reasons: 失败原因统计字典
avg_latency_ms: 平均延迟（毫秒）
"""
total_validations: int = 0
passed_validations: int = 0
failed_validations: int = 0
timeout_count: int = 0
failure_reasons: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
avg_latency_ms: float = 0.0
def get_stats(self) -> Dict[str, Any]:
"""获取统计信息
Returns:
包含验证统计、通过率、失败原因分布和平均延迟的字典
"""
pass_rate = (
self.passed_validations / self.total_validations * 100
if self.total_validations > 0 else 0
)
return {
"total_validations": self.total_validations,
"passed_validations": self.passed_validations,
"failed_validations": self.failed_validations,
"timeout_count": self.timeout_count,
"pass_rate": round(pass_rate, 2),
"failure_reasons": dict(self.failure_reasons),
"avg_latency_ms": round(self.avg_latency_ms, 2)
}
class ValidatorAgent:
"""验证Agent：审核工具调用的参数和结果是否正确
不依赖具体的LLM实现，通过协议接口与LLM客户端交互
支持重试机制、超时控制、失败原因统计和性能指标统计
Attributes:
client: 实现了LLMClientProtocol的客户端实例
config: 验证配置对象
metrics: 性能指标统计对象
"""
def __init__(self, client: LLMClientProtocol, config: ValidatorConfig):
"""初始化验证Agent
Args:
client: 实现了LLMClientProtocol的客户端实例
config: 验证配置对象
"""
self.client = client
self.config = config
self.metrics = ValidatorMetrics()
logger.info(f"ValidatorAgent 初始化完成 - model={config.model}, max_retries={config.max_retries}")
async def __aenter__(self):
"""异步上下文管理器入口
Returns:
返回自身实例，支持async with语法
"""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器出口
Args:
exc_type: 异常类型
exc_val: 异常值
exc_tb: 异常追踪
"""
await self.close()
def _build_prompt(
self,
user_message: str,
tool_name: str,
tool_type: str,
tool_args: Dict[str, Any],
tool_result: str
) -> str:
"""构建验证提示词
Args:
user_message: 用户输入的消息
tool_name: 工具名称
tool_type: 工具类型
tool_args: 工具调用参数
tool_result: 工具执行结果
Returns:
构建好的提示词
"""
return self.config.validator_prompt_template.format(
tool_type=tool_type,
tool_name=tool_name,
user_message=user_message,
tool_args=tool_args,
tool_result=tool_result
)
async def validate(
self,
user_message: str,
tool_name: str,
tool_args: Dict[str, Any],
tool_type: str,
tool_result: str
) -> Tuple[bool, str]:
"""验证工具调用的参数和结果是否正确
Args:
user_message: 用户输入的消息
tool_name: 工具名称
tool_args: 工具调用参数
tool_type: 工具类型 (WEATHER/CALCULATOR)
tool_result: 工具执行结果
Returns:
Tuple[bool, str]: (是否通过验证, 验证反馈信息)
Raises:
RuntimeError: 当重试次数耗尽时抛出
"""
self.metrics.total_validations += 1
start_time = perf_counter()
logger.info(
f"🔍 开始验证 - tool_type={tool_type}, tool_name={tool_name}, "
f"user_msg={user_message[:30]}..."
)
prompt = self._build_prompt(user_message, tool_name, tool_type, tool_args, tool_result)
messages = [{"role": "user", "content": prompt}]
# 带重试机制的执行
last_error = None
for attempt in range(self.config.max_retries):
try:
response = await asyncio.wait_for(
self.client.chat(messages=messages, tools=[]),
timeout=self.config.timeout
)
content = getattr(response, 'content', '').strip()
# 解析验证结果
if "PASS" in content.upper():
self.metrics.passed_validations += 1
self._update_latency(start_time)
logger.info(f"✅ 验证通过 - tool_type={tool_type}, attempts={attempt + 1}")
return True, "PASS"
else:
# 提取失败原因（去除 FAIL: 前缀）
feedback = content
if content.upper().startswith("FAIL:"):
feedback = content[5:].strip()
# 统计失败原因（取前50个字符作为key）
reason_key = feedback[:50]
self.metrics.failure_reasons[reason_key] += 1
self.metrics.failed_validations += 1
self._update_latency(start_time)
logger.warning(
f"⚠️ 验证未通过 - tool_type={tool_type}, "
f"reason={feedback[:50]}..., attempts={attempt + 1}"
)
return False, content
except asyncio.TimeoutError:
self.metrics.timeout_count += 1
last_error = f"验证超时 ({self.config.timeout}s)"
logger.error(f"❌ {last_error}, 尝试 {attempt + 1}/{self.config.max_retries}")
except Exception as e:
last_error = str(e)
logger.error(f"❌ 验证失败: {e}, 尝试 {attempt + 1}/{self.config.max_retries}", exc_info=True)
# 所有重试都失败
self.metrics.failed_validations += 1
logger.error(f"❌ 验证失败，重试次数耗尽: {last_error}")
raise RuntimeError(f"验证失败: {last_error}")
def _update_latency(self, start_time: float):
"""更新延迟指标
Args:
start_time: 开始时间戳
"""
latency = (perf_counter() - start_time) * 1000
self.metrics.avg_latency_ms = (
(self.metrics.avg_latency_ms * (self.metrics.total_validations - 1) + latency)
/ self.metrics.total_validations
)
def get_metrics(self) -> Dict[str, Any]:
"""获取性能指标统计
Returns:
包含各种性能指标的字典
"""
return self.metrics.get_stats()
async def close(self):
"""清理资源并关闭Agent
Note:
会关闭客户端连接
"""
if hasattr(self.client, 'close'):
await self.client.close()
logger.info("ValidatorAgent 已关闭")
def create_validator_agent(
base_url: str = "http://localhost:11434",
config: Optional[ValidatorConfig] = None
) -> ValidatorAgent:
"""创建配置好的ValidatorAgent实例
Args:
base_url: Ollama服务的基础URL
config: 可选的配置对象，未提供时使用默认配置
Returns:
配置完成的ValidatorAgent实例
Examples:
>>> agent = create_validator_agent()
>>> passed, feedback = await agent.validate("兰州天气", "tool_get_weather",
... {"city": "兰州"}, "WEATHER", "晴，25度")
>>> print(f"通过: {passed}, 反馈: {feedback}")
使用自定义配置:
>>> config = ValidatorConfig(model="llama3.1:70b", max_retries=5)
>>> agent = create_validator_agent(config=config)
"""
from app.core.pers_ollama import OllamaClient
final_config = config or ValidatorConfig()
client = OllamaClient(
base_url=base_url,
model=final_config.model,
temperature=final_config.temperature
)
return ValidatorAgent(client=client, config=final_config)
执行示例
"""
agent_validator 独立使用示例
本模块演示如何单独使用 ValidatorAgent 进行工具调用验证。
不依赖其他组件，直接传入验证所需的参数。
"""
import asyncio
import json
import logging
from app.ai_agent.agent_validator import create_validator_agent, ValidatorConfig
async def example_basic_validation():
"""示例1：基础验证 - 正确的城市名"""
print("=" * 50)
print("示例1：基础验证 - 正确的城市名")
print("=" * 50 + "\n")
async with create_validator_agent() as validator:
user_message = "兰州今天天气怎么样？"
tool_name = "tool_get_weather"
tool_type = "WEATHER"
tool_args = {"city": "兰州"}
tool_result = "晴，温度 25°C，湿度 45%"
print(f"👤 用户: {user_message}")
print(f"🛠️ 工具: {tool_name}")
print(f"📝 参数: {tool_args}")
print(f"📊 工具结果: {tool_result}\n")
passed, feedback = await validator.validate(
user_message=user_message,
tool_name=tool_name,
tool_args=tool_args,
tool_type=tool_type,
tool_result=tool_result
)
print(f"🔍 验证结果: {'✅ 通过' if passed else '❌ 未通过'}")
print(f"💬 反馈: {feedback}")
print("\n" + "-" * 50 + "\n")
async def example_wrong_city_name():
"""示例2：城市名错误验证"""
print("=" * 50)
print("示例2：城市名错误验证")
print("=" * 50 + "\n")
async with create_validator_agent() as validator:
user_message = "上海的天气怎么样？"
tool_name = "tool_get_weather"
tool_type = "WEATHER"
tool_args = {"city": "上晦"} # 错误的城市名
tool_result = "无法获取天气信息"
print(f"👤 用户: {user_message}")
print(f"🛠️ 工具: {tool_name}")
print(f"📝 参数: {tool_args}")
print(f"📊 工具结果: {tool_result}\n")
passed, feedback = await validator.validate(
user_message=user_message,
tool_name=tool_name,
tool_args=tool_args,
tool_type=tool_type,
tool_result=tool_result
)
print(f"🔍 验证结果: {'✅ 通过' if passed else '❌ 未通过'}")
print(f"💬 反馈: {feedback}")
print("\n" + "-" * 50 + "\n")
async def example_expression_validation():
"""示例3：计算表达式验证"""
print("=" * 50)
print("示例3：计算表达式验证 - 包含特殊字符")
print("=" * 50 + "\n")
async with create_validator_agent() as validator:
# 测试用例1：包含π的表达式
print("测试用例 1: 包含特殊字符 π")
user_message = "计算半径为3的圆面积"
tool_name = "tool_calculator"
tool_type = "CALCULATOR"
tool_args = {"expression": "π*3^2"}
tool_result = "表达式无效或包含非法操作: π"
print(f"👤 用户: {user_message}")
print(f"📝 参数: {tool_args}")
print(f"📊 工具结果: {tool_result}\n")
passed, feedback = await validator.validate(
user_message=user_message,
tool_name=tool_name,
tool_args=tool_args,
tool_type=tool_type,
tool_result=tool_result
)
print(f"🔍 验证结果: {'✅ 通过' if passed else '❌ 未通过'}")
print(f"💬 反馈: {feedback}\n")
print("-" * 30 + "\n")
# 测试用例2：正确的表达式
print("测试用例 2: 正确的数值表达式")
tool_args = {"expression": "3.14159 * 3 * 3"}
tool_result = "28.27431"
print(f"👤 用户: {user_message}")
print(f"📝 参数: {tool_args}")
print(f"📊 工具结果: {tool_result}\n")
passed, feedback = await validator.validate(
user_message=user_message,
tool_name=tool_name,
tool_args=tool_args,
tool_type=tool_type,
tool_result=tool_result
)
print(f"🔍 验证结果: {'✅ 通过' if passed else '❌ 未通过'}")
print(f"💬 反馈: {feedback}")
print("\n" + "-" * 50 + "\n")
async def example_batch_validation():
"""示例4：批量验证"""
print("=" * 50)
print("示例4：批量验证")
print("=" * 50 + "\n")
async with create_validator_agent() as validator:
# 定义多个验证任务
validation_cases = [
{
"description": "正确的天气查询",
"user_message": "深圳今天天气如何",
"tool_name": "tool_get_weather",
"tool_type": "WEATHER",
"tool_args": {"city": "深圳"},
"tool_result": "多云，28°C"
},
{
"description": "错误的城市名",
"user_message": "广州天气怎么样",
"tool_name": "tool_get_weather",
"tool_type": "WEATHER",
"tool_args": {"city": "广卅"},
"tool_result": "未找到城市: 广卅"
},
{
"description": "正确的计算",
"user_message": "计算 123 + 456",
"tool_name": "tool_calculator",
"tool_type": "CALCULATOR",
"tool_args": {"expression": "123 + 456"},
"tool_result": "579"
},
{
"description": "包含特殊字符e",
"user_message": "计算自然对数的底数e的近似值",
"tool_name": "tool_calculator",
"tool_type": "CALCULATOR",
"tool_args": {"expression": "e"},
"tool_result": "表达式无效"
},
]
results = []
for i, case in enumerate(validation_cases, 1):
print(f"测试 {i}: {case['description']}")
print(f"👤 用户: {case['user_message']}")
print(f"📝 参数: {case['tool_args']}")
print(f"📊 结果: {case['tool_result']}")
try:
passed, feedback = await validator.validate(
user_message=case["user_message"],
tool_name=case["tool_name"],
tool_args=case["tool_args"],
tool_type=case["tool_type"],
tool_result=case["tool_result"]
)
results.append({
"description": case["description"],
"passed": passed,
"feedback": feedback
})
print(f"🔍 {'✅ 通过' if passed else '❌ 未通过'}")
print(f"💬 {feedback[:80]}{'...' if len(feedback) > 80 else ''}")
except Exception as e:
results.append({
"description": case["description"],
"passed": False,
"feedback": f"验证异常: {str(e)}"
})
print(f"❌ 验证异常: {e}")
print()
# 汇总结果
print("=" * 50)
print("📋 验证汇总")
print("=" * 50)
passed_count = sum(1 for r in results if r["passed"])
total_count = len(results)
for result in results:
status = "✅" if result["passed"] else "❌"
print(f"{status} {result['description']}")
print(f"\n总计: {passed_count}/{total_count} 通过")
print("\n" + "-" * 50 + "\n")
async def example_with_metrics():
"""示例5：获取验证性能指标"""
print("=" * 50)
print("示例5：性能指标监控")
print("=" * 50 + "\n")
async with create_validator_agent() as validator:
# 执行多个验证任务
test_cases = [
("北京天气", "tool_get_weather", {"city": "北京"}, "WEATHER", "晴，20°C"),
("北京天气", "tool_get_weather", {"city": "北冼"}, "WEATHER", "未找到城市"),
("1+1", "tool_calculator", {"expression": "1+1"}, "CALCULATOR", "2"),
("π", "tool_calculator", {"expression": "π"}, "CALCULATOR", "表达式无效"),
("天津天气", "tool_get_weather", {"city": "天津"}, "WEATHER", "阴，18°C"),
]
print("执行验证任务...\n")
for user_msg, tool_name, args, tool_type, result in test_cases:
try:
passed, feedback = await validator.validate(
user_message=user_msg,
tool_name=tool_name,
tool_args=args,
tool_type=tool_type,
tool_result=result
)
status = "✅" if passed else "❌"
print(f"{status} {user_msg}: {feedback[:50]}...")
except Exception as e:
logging.error(f"验证失败: {e}")
print("\n")
# 获取性能指标
print("=" * 50)
print("📊 性能统计")
print("=" * 50)
stats = validator.get_metrics()
print(f"总验证次数: {stats['total_validations']}")
print(f"通过次数: {stats['passed_validations']}")
print(f"失败次数: {stats['failed_validations']}")
print(f"通过率: {stats['pass_rate']}%")
print(f"超时次数: {stats['timeout_count']}")
print(f"平均延迟: {stats['avg_latency_ms']:.2f} ms")
if stats['failure_reasons']:
print(f"\n失败原因分布:")
for reason, count in stats['failure_reasons'].items():
print(f" - [{reason}]: {count} 次")
print("\n" + "-" * 50 + "\n")
async def example_custom_config():
"""示例6：自定义配置"""
print("=" * 50)
print("示例6：自定义配置")
print("=" * 50 + "\n")
# 自定义配置
config = ValidatorConfig(
model="llama3.1:8b",
temperature=0.1, # 降低温度以获得更稳定的判断
timeout=20.0, # 增加超时时间
max_retries=2 # 减少重试次数
)
print(f"配置信息:")
print(f" - Model: {config.model}")
print(f" - Temperature: {config.temperature}")
print(f" - Timeout: {config.timeout}s")
print(f" - Max Retries: {config.max_retries}\n")
async with create_validator_agent(config=config) as validator:
user_message = "杭州天气如何"
tool_name = "tool_get_weather"
tool_type = "WEATHER"
tool_args = {"city": "杭州"}
tool_result = "小雨，22°C"
print(f"👤 用户: {user_message}\n")
passed, feedback = await validator.validate(
user_message=user_message,
tool_name=tool_name,
tool_args=tool_args,
tool_type=tool_type,
tool_result=tool_result
)
print(f"🔍 验证结果: {'✅ 通过' if passed else '❌ 未通过'}")
print(f"💬 反馈: {feedback}")
print("\n" + "-" * 50 + "\n")
async def example_timeout_handling():
"""示例7：超时处理"""
print("=" * 50)
print("示例7：超时处理（配置短超时）")
print("=" * 50 + "\n")
# 使用非常短的超时配置
config = ValidatorConfig(
model="llama3.1:8b",
temperature=0.2,
timeout=1.0, # 1秒超时
max_retries=2
)
print(f"配置超时: {config.timeout}秒\n")
async with create_validator_agent(config=config) as validator:
user_message = "成都天气怎么样"
tool_name = "tool_get_weather"
tool_type = "WEATHER"
tool_args = {"city": "成都"}
tool_result = "多云，25°C"
try:
print(f"👤 用户: {user_message}")
passed, feedback = await validator.validate(
user_message=user_message,
tool_name=tool_name,
tool_args=tool_args,
tool_type=tool_type,
tool_result=tool_result
)
print(f"🔍 验证结果: {'✅ 通过' if passed else '❌ 未通过'}")
print(f"💬 反馈: {feedback}")
except RuntimeError as e:
print(f"⏰ 验证失败（超时或重试耗尽）: {e}")
# 查看指标中的超时统计
stats = validator.get_metrics()
print(f"\n超时次数: {stats['timeout_count']}")
print("\n" + "-" * 50 + "\n")
async def main():
"""运行所有示例"""
print("\n" + "🔍" * 25)
print("ValidatorAgent 独立使用示例集合")
print("🔍" * 25 + "\n")
try:
await example_basic_validation()
await asyncio.sleep(0.5)
await example_wrong_city_name()
await asyncio.sleep(0.5)
await example_expression_validation()
await asyncio.sleep(0.5)
await example_batch_validation()
await asyncio.sleep(0.5)
await example_with_metrics()
await asyncio.sleep(0.5)
await example_custom_config()
await asyncio.sleep(0.5)
await example_timeout_handling()
except Exception as e:
logging.error(f"程序运行出错: {e}", exc_info=True)
if __name__ == "__main__":
asyncio.run(main())
执行结果
🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍
ValidatorAgent 独立使用示例集合
🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍🔍
==================================================
示例1：基础验证 - 正确的城市名
==================================================
👤 用户: 兰州今天天气怎么样？
🛠️ 工具: tool_get_weather
📝 参数: {'city': '兰州'}
📊 工具结果: 晴，温度 25°C，湿度 45%
🔍 验证结果: ✅ 通过
💬 反馈: PASS
--------------------------------------------------
==================================================
示例2：城市名错误验证
==================================================
👤 用户: 上海的天气怎么样？
🛠️ 工具: tool_get_weather
📝 参数: {'city': '上晦'}
📊 工具结果: 无法获取天气信息
⚠️ 验证未通过 - tool_type=WEATHER, reason=参数城市名 '上晦' 错误，请修正为 '上海'。..., attempts=1
🔍 验证结果: ❌ 未通过
💬 反馈: FAIL: 参数城市名 '上晦' 错误，请修正为 '上海'。
--------------------------------------------------
==================================================
示例3：计算表达式验证 - 包含特殊字符
==================================================
测试用例 1: 包含特殊字符 π
👤 用户: 计算半径为3的圆面积
📝 参数: {'expression': 'π*3^2'}
📊 工具结果: 表达式无效或包含非法操作: π
⚠️ 验证未通过 - tool_type=CALCULATOR, reason=根据重要规则，表达式处理需要检查计算器工具的表达式是否只包含数字和基本运算符，不包含特殊字符如 π、..., attempts=1
🔍 验证结果: ❌ 未通过
💬 反馈: 根据重要规则，表达式处理需要检查计算器工具的表达式是否只包含数字和基本运算符，不包含特殊字符如 π、e 等。
用户问题：计算半径为3的圆面积
参数：{'expression': 'π*3^2'}
工具结果：表达式无效或包含非法操作: π
由于表达式中包含特殊字符 π，需要转换为数值。正确的表达式应该是 '3.14159 * 3 * 3'。
因此，输出结果为：
FAIL: 表达式非法，转化为纯数字的公式，不使用特殊字符，例如'3.14159 * 3 * 3'。
------------------------------
测试用例 2: 正确的数值表达式
👤 用户: 计算半径为3的圆面积
📝 参数: {'expression': '3.14159 * 3 * 3'}
📊 工具结果: 28.27431
🔍 验证结果: ✅ 通过
💬 反馈: PASS
--------------------------------------------------
==================================================
示例4：批量验证
==================================================
测试 1: 正确的天气查询
👤 用户: 深圳今天天气如何
📝 参数: {'city': '深圳'}
📊 结果: 多云，28°C
🔍 ✅ 通过
💬 PASS
测试 2: 错误的城市名
👤 用户: 广州天气怎么样
📝 参数: {'city': '广卅'}
📊 结果: 未找到城市: 广卅
⚠️ 验证未通过 - tool_type=WEATHER, reason=工具未找到城市 '广卅'。用户问的是 '广州'，请直接使用 '广州' 作为参数。..., attempts=1
🔍 ❌ 未通过
💬 FAIL: 工具未找到城市 '广卅'。用户问的是 '广州'，请直接使用 '广州' 作为参数。
测试 3: 正确的计算
👤 用户: 计算 123 + 456
📝 参数: {'expression': '123 + 456'}
📊 结果: 579
🔍 ✅ 通过
💬 PASS
测试 4: 包含特殊字符e
👤 用户: 计算自然对数的底数e的近似值
📝 参数: {'expression': 'e'}
📊 结果: 表达式无效
⚠️ 验证未通过 - tool_type=CALCULATOR, reason=根据重要规则，计算器工具的表达式必须只包含数字和基本运算符，不能包含特殊字符，如 π、e 等。用户的..., attempts=1
🔍 ❌ 未通过
💬 根据重要规则，计算器工具的表达式必须只包含数字和基本运算符，不能包含特殊字符，如 π、e 等。用户的问题中提到了自然对数的底数 e，这意味着表达式中存在特殊字符...
==================================================
📋 验证汇总
==================================================
✅ 正确的天气查询
❌ 错误的城市名
✅ 正确的计算
❌ 包含特殊字符e
总计: 2/4 通过
--------------------------------------------------
==================================================
示例5：性能指标监控
==================================================
执行验证任务...
✅ 北京天气: PASS...
⚠️ 验证未通过 - tool_type=WEATHER, reason=工具未找到城市 '北冼'。用户问的是 '北京'，请直接使用 '北京' 作为参数。..., attempts=1
❌ 北京天气: FAIL: 工具未找到城市 '北冼'。用户问的是 '北京'，请直接使用 '北京' 作为参数。...
✅ 1+1: PASS...
⚠️ 验证未通过 - tool_type=CALCULATOR, reason=根据重要规则和示例，我们可以得出以下结论：
1. 用户问题： π
2. 工具参数：{'expres..., attempts=1
❌ π: 根据重要规则和示例，我们可以得出以下结论：
1. 用户问题： π
2. 工具参数：{'expres...
✅ 天津天气: PASS...
==================================================
📊 性能统计
==================================================
总验证次数: 5
通过次数: 3
失败次数: 2
通过率: 60.0%
超时次数: 0
平均延迟: 5093.14 ms
失败原因分布:
- [工具未找到城市 '北冼'。用户问的是 '北京'，请直接使用 '北京' 作为参数。]: 1 次
- [根据重要规则和示例，我们可以得出以下结论：
1. 用户问题： π
2. 工具参数：{'expres]: 1 次
--------------------------------------------------
==================================================
示例6：自定义配置
==================================================
配置信息:
- Model: llama3.1:8b
- Temperature: 0.1
- Timeout: 20.0s
- Max Retries: 2
👤 用户: 杭州天气如何
🔍 验证结果: ✅ 通过
💬 反馈: PASS
--------------------------------------------------
==================================================
示例7：超时处理（配置短超时）
==================================================
配置超时: 1.0秒
👤 用户: 成都天气怎么样
❌ 验证超时 (1.0s), 尝试 1/2
❌ 验证超时 (1.0s), 尝试 2/2
❌ 验证失败，重试次数耗尽: 验证超时 (1.0s)
⏰ 验证失败（超时或重试耗尽）: 验证失败: 验证超时 (1.0s)
超时次数: 2
--------------------------------------------------
以上便是本文的全部内容，感谢您的阅读，如遇到任何问题，欢迎在评论区留言讨论。