构建可验证、可评测的AI Agent工具调用工程框架-北京尧图网络

1. 项目概述从“玩具”到“工程”的跨越最近和几个做AI应用的朋友聊天发现一个挺普遍的现象大家用LangChain、AutoGPT或者OpenAI的Assistant API搭个能聊天的Agent跑通一个Demo感觉挺酷但一到要真正上线或者给客户演示问题就全来了。工具调用时灵时不灵返回的JSON格式说变就变同一个问题这次能成功调用天气API下次就给你返回一堆胡言乱语。这让我意识到我们很多人还停留在“会聊天的玩具”阶段离一个稳定、可靠、可评估的“工程系统”还差得远。我这个项目就是被这种不稳定性给“逼”出来的。核心目标很明确用一天时间构建一个不仅能让Agent调用工具还能对每次调用进行验证、评测和追踪的轻量级工程框架。我不想再造一个庞大的Agent框架而是想做一个“脚手架”或者“中间件”它能无缝嵌入到你现有的Agent工作流里把工具调用这个黑盒过程变成白盒的、可观测的、可度量的工程环节。关键词是“可验证”和“可评测”——这意味着每次工具调用前我们能预判其合理性调用后我们能评估其有效性。这适合谁呢如果你正在将LLM驱动的自动化流程比如智能客服、数据分析助手、自动化运维机器人从原型推向生产或者你受够了Agent在演示时“掉链子”需要一套机制来保证其行为的一致性和可靠性那么这个思路和接下来的实现细节应该能给你带来不少启发。我们最终要的不是一个只会聊天的AI而是一个能踏实干活的、值得信赖的“智能员工”。2. 核心设计思路为工具调用加上“质检流水线”传统的Agent工具调用流程大致是“LLM生成调用指令 - 解析指令 - 执行工具 - 返回结果给LLM”。这个流程的问题在于它假设LLM每次生成的指令都是完美且上下文恰当的但现实往往骨感。我的设计思路是在这个流程中插入两个关键的“质检站”形成一个可验证、可评测的闭环系统。2.1 架构总览三层质检闭环整个系统的核心是一个三层处理流水线我把它叫做“调用生命周期管理”。意图验证层调用前在LLM生成工具调用请求后、实际执行前对请求进行拦截和校验。检查内容包括请求的格式是否符合工具定义的Schema传入的参数值是否在合理范围内比如查询天气的“城市”参数不能是“火星”本次调用在当前对话上下文中是否逻辑通顺比如用户刚问完北京天气紧接着Agent不应该去调用计算器。执行监控层调用中工具执行本身可能出错网络超时、API限流、资源不存在。这一层负责封装工具调用加入重试机制、超时控制、异常捕获和详细的日志记录确保执行过程的健壮性和可观测性。结果评测层调用后工具成功返回结果但这结果就是好的吗这一层对返回结果进行“质检”。例如调用搜索引擎工具返回了摘要需要评估摘要是否真正回答了用户问题是否包含无关或错误信息。这里会引入一套轻量级的评测规则或模型。这个三层架构相当于给工具调用套上了一个“标准化作业流程”SOP每一步都有检查点和记录从而将随机的、不稳定的行为转变为可预测、可复盘的过程。2.2 技术选型轻量、灵活、即插即用为了实现“1天搭建”的目标技术选型上必须追求极致的轻量和简洁。核心语言与框架Python是不二之选。生态丰富异步支持好与主流AI库无缝集成。我刻意避开了功能庞大但学习曲线陡峭的全栈框架选择基于Pydantic来构建数据验证的基石。Pydantic的模型定义Model能完美地描述一个工具的输入输出Schema并且自带强大的类型验证和JSON序列化能力这为我们的“意图验证层”提供了几乎零成本的实现方案。工具调用基础直接使用OpenAI的Chat Completion API的function calling或更新的tool calls能力作为起点。它们已经定义了LLM与工具交互的标准格式JSON Schema我们只需要在这个标准之上增加我们的验证和监控逻辑。异步与任务管理使用asyncio处理可能的并发工具调用对于需要重试和复杂状态管理的任务Celery或更轻量的RQ是备选但在“1天”的极限挑战下我优先采用简单的asyncio.gather配合超时控制保持核心逻辑的简洁。可观测性日志记录是系统的眼睛。使用结构化的日志如Python的structlog或json-logger将每次工具调用的请求ID、验证结果、执行耗时、返回状态、评测分数等关键信息以JSON格式输出。这方便后续用Grafana或ELK进行聚合分析和可视化。注意这里没有选择LangChain等高级框架并非它们不好而是在这个特定目标下它们引入了过多的抽象和概念不利于我们快速、透明地构建“质检流水线”。我们的系统更像是一个增强插件应该能够适配多种底层Agent实现。3. 核心模块拆解与实现接下来我们深入到代码层面看看这三个核心层是如何具体实现的。我会用最关键的代码片段来说明思路你可以根据自身需求调整。3.1 模块一基于Pydantic的工具定义与意图验证器这是系统的基石。我们首先要标准化工具的描述。from pydantic import BaseModel, Field, validator from typing import Optional, Any, Callable from enum import Enum class ToolResultStatus(str, Enum): SUCCESS success VALIDATION_ERROR validation_error EXECUTION_ERROR execution_error CONTEXT_ERROR context_error class ToolDefinition(BaseModel): 工具定义模型 name: str Field(..., description工具的唯一名称) description: str Field(..., description工具功能的自然语言描述) input_schema: type[BaseModel] Field(..., description输入参数的Pydantic模型) func: Callable Field(..., description实际执行的函数) validator_hooks: Optional[list[Callable]] Field(default_factorylist, description自定义验证钩子) class ValidatedToolCall(BaseModel): 经过验证的工具调用请求 tool_name: str arguments: dict[str, Any] status: ToolResultStatus message: Optional[str] None raw_request: dict[str, Any] Field(excludeTrue) # 保存原始请求用于调试 class ToolValidator: 工具验证器 def __init__(self, tool_registry: dict[str, ToolDefinition]): self.tools tool_registry async def validate_call(self, llm_raw_tool_call: dict) - ValidatedToolCall: 验证LLM原始工具调用请求。 1. 格式与基础校验 2. 参数值域校验 3. 自定义钩子校验如上下文逻辑 validated ValidatedToolCall( tool_namellm_raw_tool_call.get(name, ), argumentsllm_raw_tool_call.get(arguments, {}), raw_requestllm_raw_tool_call, statusToolResultStatus.SUCCESS ) # 1. 工具是否存在 if validated.tool_name not in self.tools: validated.status ToolResultStatus.VALIDATION_ERROR validated.message f工具 {validated.tool_name} 未注册。 return validated tool_def self.tools[validated.tool_name] # 2. 参数Schema校验 (利用Pydantic) try: # 这里会触发Pydantic的字段类型、必填项等基础验证 parsed_args tool_def.input_schema(**validated.arguments) validated.arguments parsed_args.dict() # 转换为经过验证的字典 except Exception as e: validated.status ToolResultStatus.VALIDATION_ERROR validated.message f参数校验失败: {str(e)} return validated # 3. 执行自定义验证钩子例如检查参数业务逻辑 for hook in tool_def.validator_hooks: hook_result await hook(validated.tool_name, validated.arguments, self.context) # context为当前会话上下文 if not hook_result.is_valid: validated.status ToolResultStatus.CONTEXT_ERROR validated.message hook_result.error_message return validated return validated实操要点ToolDefinition中的input_schema必须是一个Pydantic.BaseModel。例如一个“获取天气”工具其input_schema可以定义为WeatherQuery(city: str Field(..., regex^[a-zA-Z\s]$), country: str Field(CN))。这样城市名只能是字母和空格国家代码默认为‘CN’。validator_hooks是一个可扩展列表。你可以加入一个钩子函数检查“在当前对话历史里用户是否已经提供了城市信息”如果没有则让验证失败并提示Agent需要先向用户澄清。这是实现“上下文逻辑校验”的关键。验证结果ValidatedToolCall包含了明确的状态枚举这为后续流程的路由是继续执行还是直接返回错误给LLM/用户提供了清晰依据。3.2 模块二带监控的执行器与结果封装验证通过后请求进入执行层。这一层要确保执行过程的稳定并丰富结果信息。import asyncio import time from contextlib import asynccontextmanager from dataclasses import dataclass from typing import Optional dataclass class ExecutionMetrics: 执行度量指标 start_time: float end_time: Optional[float] None retry_count: int 0 error_type: Optional[str] None property def duration(self) - float: return self.end_time - self.start_time if self.end_time else 0.0 class ToolExecutor: 工具执行器 def __init__(self, max_retries: int 2, timeout: float 30.0): self.max_retries max_retries self.timeout timeout async def execute(self, validated_call: ValidatedToolCall, tool_def: ToolDefinition) - dict: 执行已验证的工具调用并封装结果。 metrics ExecutionMetrics(start_timetime.time()) result_data None final_status ToolResultStatus.SUCCESS error_msg None for attempt in range(self.max_retries 1): try: # 使用异步超时控制 async with self._timeout_context(self.timeout): # 实际执行工具函数 if asyncio.iscoroutinefunction(tool_def.func): result_data await tool_def.func(**validated_call.arguments) else: # 如果是同步函数放到线程池执行避免阻塞事件循环 result_data await asyncio.to_thread(tool_def.func, **validated_call.arguments) metrics.end_time time.time() break # 成功则跳出重试循环 except asyncio.TimeoutError: metrics.error_type timeout error_msg f工具执行超时{self.timeout}s if attempt self.max_retries: final_status ToolResultStatus.EXECUTION_ERROR except Exception as e: metrics.error_type type(e).__name__ error_msg f执行异常: {str(e)} if attempt self.max_retries: final_status ToolResultStatus.EXECUTION_ERROR await asyncio.sleep(1 * (attempt 1)) # 简单的指数退避 metrics.retry_count attempt 1 # 结构化返回结果 return { request_id: validated_call.raw_request.get(id), # 关联原始请求 tool_name: validated_call.tool_name, status: final_status, data: result_data, error_message: error_msg, metrics: { duration_seconds: round(metrics.duration, 3), retry_count: metrics.retry_count, error_type: metrics.error_type }, validated_arguments: validated_call.arguments # 记录实际使用的参数 } asynccontextmanager async def _timeout_context(self, timeout): 异步超时上下文管理器 try: yield except asyncio.CancelledError: raise asyncio.TimeoutError()实操心得超时控制是生命线网络工具调用必须设置超时否则一个挂死的请求会拖垮整个Agent。asyncio.wait_for或自定义的上下文管理器是标准做法。区分同步与异步函数你的工具函数可能是访问数据库的同步库也可能是调用异步HTTP客户端的异步函数。执行器需要能透明地处理这两种情况asyncio.to_thread()是处理CPU密集型或阻塞式同步调用的好帮手。度量指标Metrics是黄金记录下的duration和retry_count是后续进行性能分析和可靠性评估的核心数据。哪个工具最慢哪个工具失败率最高一目了然。3.3 模块三结果评测器与反馈循环工具执行成功了返回了数据但这数据“质量”如何我们需要一个评测机制。class ResultEvaluator: 结果评测器 def __init__(self, evaluation_rules: dict): evaluation_rules: 一个字典key为工具名value为该工具对应的评测函数列表。例如{“get_weather”: [self._check_data_completeness, self._check_relevance_to_query]} self.rules evaluation_rules async def evaluate(self, tool_name: str, result: dict, original_user_query: str, context: dict) - dict: 对工具执行结果进行多维度评测。返回一个包含各项分数和综合评估的字典。 if tool_name not in self.rules: return {overall_score: 1.0, details: {}, passed: True} # 默认通过 scores {} details {} all_passed True for rule_func in self.rules[tool_name]: rule_name rule_func.__name__ try: # 每个评测规则返回一个分数0-1和详情 score, detail await rule_func(result[data], original_user_query, context) scores[rule_name] score details[rule_name] detail if score 0.6: # 假设阈值是0.6 all_passed False except Exception as e: # 评测规则本身出错不应导致系统崩溃 scores[rule_name] 0.0 details[rule_name] f评测规则执行失败: {e} all_passed False # 计算综合分数例如取平均值 overall_score sum(scores.values()) / len(scores) if scores else 1.0 evaluation_result { overall_score: overall_score, score_breakdown: scores, details: details, passed: all_passed and overall_score 0.6 } # 将评测结果附加到原始结果上形成增强结果 result[evaluation] evaluation_result return result # 示例评测规则1数据完整性检查 async def _check_data_completeness(self, data: dict, query: str, context: dict) - tuple[float, str]: 检查返回的数据是否包含所有必要字段。 required_fields [temperature, weather_condition, humidity] missing [f for f in required_fields if f not in data] score 1.0 if not missing else 0.3 detail f缺失字段: {missing} if missing else 所有必要字段完整。 return score, detail # 示例评测规则2结果相关性检查可简化为关键词匹配复杂情况可调用小模型 async def _check_relevance_to_query(self, data: dict, query: str, context: dict) - tuple[float, str]: 简单检查返回的文本摘要是否包含查询中的关键词。 summary data.get(summary, ) keywords set(query.lower().split()[:5]) # 取前5个词作为关键词 found_keywords [kw for kw in keywords if kw in summary.lower()] relevance_ratio len(found_keywords) / len(keywords) if keywords else 1.0 detail f查询关键词命中率: {relevance_ratio:.2f} ({found_keywords}) return relevance_ratio, detail设计考量规则化与可配置评测器不是硬编码的而是通过规则字典配置。你可以为不同的工具轻松添加、移除或修改评测规则实现了关注点分离。轻量级优先最初的评测规则应该简单、快速比如字段检查、格式验证、关键词匹配。这足以过滤掉大部分明显的垃圾结果。只有当简单规则不够用时再考虑引入更复杂的基于嵌入向量的相似度计算甚至调用一个小型LLM如GPT-3.5-Turbo来评判结果质量。记住评测本身不应该成为性能瓶颈。反馈循环评测结果尤其是“未通过”的结果不应该被默默丢弃。它可以有几种用途1) 直接作为错误信息返回给LLM让其重新思考或调整问题2) 记录到日志中用于后续的集中分析和工具优化3) 触发一个告警通知开发者某个工具最近返回低质量结果。4. 系统集成与工作流编排现在我们把验证器、执行器、评测器像乐高积木一样组装起来形成一个完整的工作流。这个工作流应该能轻松嵌入到现有的Agent主循环中。4.1 核心协调器ToolManagerclass ToolManager: 工具管理协调器串联验证、执行、评测全流程 def __init__(self, validator: ToolValidator, executor: ToolExecutor, evaluator: ResultEvaluator): self.validator validator self.executor executor self.evaluator evaluator self.logger structlog.get_logger(__name__) async def process_tool_calls(self, raw_tool_calls: list[dict], user_query: str, session_context: dict) - list[dict]: 处理一批原始工具调用请求。这是对外的主要接口。 final_results [] for raw_call in raw_tool_calls: call_id raw_call.get(id, unknown) self.logger.info(tool_call_started, call_idcall_id, tool_nameraw_call.get(name)) # 1. 验证 validated await self.validator.validate_call(raw_call) if validated.status ! ToolResultStatus.SUCCESS: self.logger.warning(tool_call_validation_failed, call_idcall_id, statusvalidated.status, messagevalidated.message) final_results.append({ call_id: call_id, status: validated.status.value, error: validated.message, data: None }) continue # 验证失败跳过执行 # 2. 执行 tool_def self.validator.tools[validated.tool_name] execution_result await self.executor.execute(validated, tool_def) if execution_result[status] ! ToolResultStatus.SUCCESS: self.logger.error(tool_call_execution_failed, call_idcall_id, errorexecution_result[error_message]) final_results.append(execution_result) continue # 执行失败跳过评测 # 3. 评测 evaluated_result await self.evaluator.evaluate( validated.tool_name, execution_result, user_query, session_context ) # 记录成功日志包含度量指标和评测分数 self.logger.info(tool_call_completed, call_idcall_id, tool_namevalidated.tool_name, durationevaluated_result[metrics][duration_seconds], eval_scoreevaluated_result.get(evaluation, {}).get(overall_score, 1.0), eval_passedevaluated_result.get(evaluation, {}).get(passed, True) ) final_results.append(evaluated_result) return final_results4.2 与现有Agent框架集成这个ToolManager的设计是松耦合的。你可以很容易地将它集成到不同的Agent框架中。场景一集成到自定义的Agent循环中# 你的主Agent循环 async def agent_loop(user_input: str, history: list): # 1. 调用LLM获取包含tool_calls的响应 llm_response await openai_client.chat.completions.create( modelgpt-4, messageshistory [{role: user, content: user_input}], toolstool_definitions_for_openai, # 将你的ToolDefinition转换成OpenAI格式 tool_choiceauto ) # 2. 检查是否有工具调用 if tool_calls : llm_response.choices[0].message.tool_calls: # 3. 交给我们的ToolManager处理 tool_results await tool_manager.process_tool_calls( raw_tool_calls[tc.model_dump() for tc in tool_calls], user_queryuser_input, session_context{history: history} ) # 4. 将处理结果构建成LLM需要的格式继续对话 new_messages [] for result in tool_results: if result[status] success: # 成功的结果构造tool message new_messages.append({ role: tool, content: str(result[data]), # 或者经过格式化的内容 tool_call_id: result[request_id] }) else: # 失败的结果可以构造一个错误信息让LLM知晓 new_messages.append({ role: tool, content: fTool call failed: {result.get(error_message)}, tool_call_id: result[request_id] }) history.extend(new_messages) # 继续循环将包含工具结果的历史再次发给LLM...场景二作为LangChain的自定义Tool装饰器你可以将ToolManager包装成一个LangChain的Custom Tool利用其强大的生态进行编排同时享受我们系统的验证和监控能力。集成心得保持接口简单ToolManager.process_tool_calls的输入输出都是简单的字典列表这使其能适配几乎所有框架。上下文传递注意将session_context如对话历史、用户ID一路传递下去这对于验证层和评测层的规则判断至关重要。错误处理策略当工具调用失败或评测不通过时你需要决定是让Agent直接向用户报错还是让LLM根据错误信息尝试其他策略比如换一个工具或者向用户追问。我们的系统提供了清晰的错误状态和原因把决策权交还给上层的Agent策略。5. 可观测性与评测体系构建系统跑起来了但怎么知道它好不好我们需要建立可观测性和一套评测体系。5.1 结构化日志与监控看板我们已经在代码中关键节点插入了结构化日志。下一步是将这些日志收集起来并可视化。日志字段标准化确保每个工具调用日志都包含call_id,tool_name,phasevalidation/execution/evaluation,status,duration,error_message,eval_score等核心字段。使用日志聚合系统将日志发送到Loki或ELKElasticsearch, Logstash, Kibana栈。这样你可以轻松查询“过去一小时get_stock_price工具的平均执行时间是多少”、“search_web工具的验证失败率有没有突然升高”构建Grafana看板这是直观掌握系统健康度的关键。可以创建几个核心面板吞吐量与延迟面板显示各工具每分钟调用次数、平均/百分位延迟P50, P95, P99。错误率面板按工具、按错误类型验证错误、执行超时、API错误统计失败率。结果质量面板展示各工具返回结果的评测平均分分布以及“未通过”评测的样例。5.2 设计有效的评测指标除了系统层面的监控我们还需要业务层面的评测指标来衡量工具调用是否真的“有用”。评测维度具体指标测量方法目标可靠性调用成功率(成功调用次数) / (总调用次数) 99.5%平均故障间隔MTBF统计连续成功调用之间的平均时间尽可能长性能平均响应时间从接收到请求到返回结果的平均耗时根据工具类型设定SLA如2s尾延迟P99最慢的1%请求的耗时控制在一定阈值内避免极端体验有效性结果相关度得分通过评测器中的规则计算出的平均分 0.8用户任务完成率在使用了工具调用的对话中最终解决用户问题的比例通过人工或模型抽样评估效率工具调用必要性在最终成功解决问题的对话中工具调用被判定为“必要”的比例避免无效调用降低成本和延迟实操心得从简单指标开始不要一开始就追求完美的“用户任务完成率”评估这需要大量标注。先从最客观的成功率和延迟开始监控它们能立刻暴露系统稳定性问题。定期进行人工评估每周随机抽取100条包含工具调用的对话日志让人快速判断“这次工具调用是否必要且正确”。这个“黄金标准”数据可以用来校准你的自动评测规则比如_check_relevance_to_query的阈值。建立反馈闭环将监控和评测中发现的问题例如某个工具频繁超时、某个参数经常被LLM填错反馈到开发流程中。是工具API需要优化还是需要给LLM提供更清晰的工具描述或者是我们的验证规则需要加强用数据驱动迭代。6. 常见问题与实战避坑指南在实际搭建和运行这套系统的过程中我踩过不少坑也总结出一些让系统更稳健的经验。6.1 验证层常见陷阱Schema变更的兼容性问题当你修改了工具的Pydantic Schema比如增加了一个可选字段之前缓存的或正在传输中的旧格式请求可能会导致验证失败。解决方案为Schema添加版本号或者在验证时采用更宽松的模式如extra‘ignore’并在日志中记录Schema不匹配的警告便于后续清理。上下文验证的复杂度检查“调用是否合乎上下文逻辑”的钩子函数可能会变得非常复杂影响性能。建议保持钩子函数轻量级。复杂的逻辑判断如基于整个对话历史做意图分析可以考虑提前计算好以特征的形式存入session_context供钩子函数快速读取。6.2 执行层稳定性保障工具函数的副作用与幂等性有些工具调用有副作用如发送邮件、创建订单。如果因为网络超时导致重试可能造成重复执行。关键设计对于非幂等的工具必须在ToolDefinition中明确标记并且执行器要禁用其重试机制。更好的做法是让工具函数自身实现幂等性例如通过唯一的业务ID来避免重复创建。资源泄漏工具函数可能打开网络连接、数据库连接或文件句柄。如果执行过程中发生异常必须确保资源被正确释放。建议在执行器内部使用try...finally块或异步上下文管理器来确保清理逻辑一定会执行。6.3 评测层的平衡艺术评测规则本身的准确性一个糟糕的评测规则可能会把好结果误判为坏结果或者放过坏结果。应对策略定期用人工评估的结果作为基准计算你的自动评测规则的准确率、召回率并持续优化它们。评测带来的额外延迟复杂的评测规则尤其是调用另一个LLM会显著增加整体响应时间。优化建议将评测设计为异步、非阻塞的。即主流程不等待评测结果就先将工具结果返回给LLM继续推理。评测在后台运行结果用于后续的分析和告警不影响本次响应的实时性。6.4 系统集成与调试日志太多找不到重点结构化日志如果字段设计不当会变得臃肿。技巧定义不同日志级别。INFO级别记录核心流程开始、结束、关键指标DEBUG级别记录详细的参数和中间结果。并利用日志聚合系统的过滤和查询功能。如何测试整个流水线为ToolManager编写单元测试和集成测试。使用pytest和pytest-asyncio。模拟LLM的原始调用、模拟工具函数的不同行为成功、抛出异常、返回特定格式的数据验证验证器、执行器、评测器是否能按预期工作。

构建可验证、可评测的AI Agent工具调用工程框架

相关推荐

【单片机课设毕设项目】基于 STM32 的嵌入式密码指纹锁安全防护装置设计基于 STM32 的多方式解锁嵌入式门禁系统设计与实现（012502）

U盘启动盘制作与Windows系统重装全流程详解

从零上手码云：Git版本控制与团队协作实战指南

更多资讯

MCP、Rule、Skill、Agent区别与关系：一张图理清AI智能体核心概念

ArcGIS XY坐标批量转点线面要素：从坐标系避坑到arcpy脚本实战

Claude Code模板工程：用CLAUDE.md与Hooks约束AI，告别随机输出

内省阈值：为什么LLM无法真正自我进化？

NodeMCU网页发中文串口显示：全链路编码与乱码排查

多参数贝叶斯估计的核心：联合概率分布与MCMC后验采样实战

今日更新

NetSpot 实战：用信道分析与频宽优化解决 Wi-Fi 卡顿

C++ split() 字符串分割的五种实现与性能实测

R语言坐标轴范围设置：从xlim/ylim到coord_cartesian的完整指南

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？