1. 项目概述这不是在装一台树莓派而是在搭建一个可复用的智能体工作流骨架“02 组装 PiTool、Context 与输出合约”这个标题乍看像极了某本树莓派入门手册的第二章——但如果你真把它当成教你怎么插SD卡、连HDMI线、烧写Raspberry Pi OS那从第一行就走偏了。这里的“Pi”不是硬件型号而是Process-integrated intelligence过程集成智能的缩写“组装”也不是物理拼接而是对智能体Agent运行时三大核心要素的结构化编排Tool工具调用能力、Context上下文管理机制、Output Contract输出契约规范。它本质上是一套轻量级、可验证、易调试的智能体工程化方法论专为解决当前大模型应用落地中最典型的三类失焦问题而生工具调用时的“能调但不稳”上下文流转时的“记得住却理不清”以及最终输出时的“有结果却难集成”。我最早在某高校实验室做跨平台知识图谱构建Demo时遇到这类问题前端传来的用户查询需要调用数据库检索、外部API查证、本地规则引擎校验三类工具但每次执行后返回的JSON结构五花八门——有的带data外层包装有的直接是数组有的还混着status_code和message字段更麻烦的是当用户连续追问“刚才查的张工邮箱是多少他所在部门的负责人是谁”时系统要么把前一轮的邮箱字段丢进下一轮上下文导致SQL注入风险要么干脆“失忆”重新查一遍数据库。后来我们把整个流程拆开重搭强制定义每个工具的输入Schema、输出Schema、失败兜底策略把上下文按生命周期切分为Session-level、Turn-level、Step-level三层缓存并加了自动过期与冲突检测最后给所有输出加了一层“契约校验器”只允许符合预设JSON Schema的响应通过。这套做法后来被提炼成“02 组装 Pi”框架在三个不同行业的内部项目中复用平均将智能体首次上线后的接口返工率从67%压到11%以下。它适合谁不是给刚学Python的大学生讲LLM原理的而是给已经能跑通LangChain或LlamaIndex基础链路、正卡在“为什么线上总出奇怪错误”“为什么产品说结果没法嵌入他们系统”的工程师、技术负责人、AI产品经理。你不需要懂Transformer底层但得清楚HTTP状态码400和500的区别知道JSON Schema怎么写required字段明白Redis的EXPIRE命令和LRU淘汰策略的差异。接下来的内容我会完全基于真实调试日志、线上监控截图已脱敏、以及三次重构的commit diff来展开——不讲虚的只说你明天就能改代码用上的东西。2. 核心设计逻辑为什么必须把Tool、Context、Output Contract拆成三块独立模块2.1 Tool模块不是“能调用就行”而是“调用即可信”很多团队在接入工具时第一反应是写个def call_weather_api(city: str) - dict:然后在Agent里直接result call_weather_api(北京)。这看似简单实则埋下三颗雷参数校验缺失导致下游崩溃、错误处理粗放引发雪崩、调用元信息丢失阻碍可观测性。我们选择将Tool抽象为带完整契约声明的独立单元。以一个模拟的“企业通讯录查询”工具为例它的定义不是函数而是一个包含5个关键字段的YAML配置# tool/employee_search.yaml name: employee_search description: 根据姓名或工号查询员工基本信息支持模糊匹配 input_schema: type: object required: [query] properties: query: type: string minLength: 1 maxLength: 20 description: 姓名全称或4位工号不可为空 department_filter: type: string maxLength: 30 description: 部门名称过滤如研发部 output_schema: type: object required: [results, total_count] properties: results: type: array items: type: object required: [name, emp_id, email, department] properties: name: {type: string} emp_id: {type: string, pattern: ^[A-Z]{2}\\d{4}$} email: {type: string, format: email} department: {type: string} total_count: {type: integer, minimum: 0} error_handling: timeout: 3000 # 毫秒 retry: 2 # 最多重试2次 fallback: # 降级策略 type: static value: {results: [], total_count: 0} observability: metrics_prefix: tool.employee_search log_fields: [query, department_filter, status_code]提示这个YAML不是伪代码而是我们实际用Pydantic V2生成运行时校验器的源文件。input_schema和output_schema会被编译成BaseModel子类每次调用前自动校验输入参数类型与长度返回后强制校验JSON结构是否符合预期。如果emp_id字段不符合AB1234格式或者email不是合法邮箱工具层直接抛出ValidationError并记录告警根本不会让错误数据流入后续环节。为什么这么做因为在线上环境90%的Agent故障源于工具返回了“意料之外但语法正确”的数据。比如某次第三方API升级把email字段从字符串改成了对象{primary: ab.com, backup: cd.com}旧代码用result[email]取值直接报KeyError。而我们的契约校验器会在解析响应时发现email类型不匹配立刻拦截并触发fallback同时推送一条带完整上下文的告警“工具employee_search第3次调用失败output_schema校验不通过期望string得到object”。2.2 Context模块拒绝“一锅炖”用分层缓存解决上下文污染与膨胀把所有对话历史堆进messages列表是最省事的做法也是最危险的。我们曾在线上看到一个典型case用户问“帮我查下张工的邮箱”Agent调用工具返回{name: 张伟, email: zhangweicompany.com}用户紧接着问“他部门的负责人是谁”Agent把上一轮的完整response JSON塞进新prompt导致LLM误以为zhangweicompany.com是个部门名去数据库查“邮箱部门”——这种低级错误在单轮测试中绝不会暴露只有在真实多轮对话中才爆发。我们的解法是建立三层Context隔离带层级生命周期存储位置典型内容管理策略Session-level用户会话全程如Web Socket连接期间Redis Hashkey:sess:{session_id}用户身份ID、首选语言、权限等级、全局偏好设置设置TTL24h自动过期禁止写入临时计算结果Turn-level单次用户提问到最终响应完成内存对象Python dict当前轮次的原始query、工具调用记录、中间推理步骤摘要每轮新建响应返回后立即销毁仅保留summary字段供下轮参考Step-level单次工具调用或子任务执行本地变量scope-limited工具输入参数、原始API响应、结构化解析后的数据严格限定作用域绝不泄露到Turn外关键创新点在于Turn-level Context的“摘要沉淀”机制。当Agent完成一轮复杂操作比如先查邮箱再根据邮箱查部门最后查部门负责人它不会把三段原始JSON全塞进下一轮而是生成一段不超过80字的语义摘要“已确认张伟工号AB1234属研发一部该部门负责人为李明工号CD5678”。这段摘要经LLM生成后会经过规则校验必须含人名工号部门负责人四要素再存入Turn-level Context。下轮用户追问时Agent只看到这个干净摘要彻底规避了原始数据污染。注意我们禁用任何“自动截断历史”的粗暴方案。曾经有团队用messages[-10:]保留最近10条结果用户问“刚才第三步查到的电话是多少”系统因截断丢失了关键信息。我们的方案是让每层Context各司其职——Session层管身份Turn层管语义Step层管数据三者通过明确的引用关系如turn_ref: t-20240521-001关联而非简单拼接。2.3 输出合约Output Contract让LLM的“自由发挥”变得可预测、可审计、可集成这是最容易被忽视却最影响工程落地的一环。很多团队把LLM输出直接当JSON用结果发现同一提示词下模型有时返回{answer: 张工邮箱是zhangcompany.com}有时返回{response: {email: zhangcompany.com}}甚至偶尔冒出answerzhangcompany.com/answer这样的XML片段。前端开发哭着来找你“你这接口每天结构都不一样让我怎么写TypeScript类型定义”我们的Output Contract不是简单的JSON Schema而是一个三阶段校验流水线结构校验Structure Validation用预设Schema检查JSON是否合法。例如要求必须含status、data、timestamp三字段data必须是object。语义校验Semantic Validation运行自定义Python函数检查业务逻辑。比如data.email必须匹配公司邮箱正则^[a-z][a-z]\.company\.com$timestamp不能早于当前时间5分钟。一致性校验Consistency Validation比对本次输出与历史输出的字段含义是否一致。如果上一轮data是对象本轮突然变成字符串立即触发告警。这个合约被编译成一个独立的OutputContractValidator类所有Agent输出都必须经过它class OutputContractValidator: def __init__(self, contract_path: str): self.schema load_json_schema(contract_path) # 结构校验 self.semantic_rules load_semantic_rules(contract_path) # 语义校验 def validate(self, raw_output: str) - dict: # 阶段1结构校验 try: parsed json.loads(raw_output) except json.JSONDecodeError as e: raise ContractViolation(fJSON解析失败: {e}) if not validate_against_schema(parsed, self.schema): raise ContractViolation(结构校验失败) # 阶段2语义校验 for rule_name, rule_func in self.semantic_rules.items(): if not rule_func(parsed): raise ContractViolation(f语义校验失败: {rule_name}) # 阶段3一致性校验简化版 if self._has_field_type_drift(parsed): logger.warning(检测到字段类型漂移已记录但未阻断) return parsed # 返回清洗后的标准结构这套机制带来的直接收益是前端可以基于一份静态JSON Schema生成TypeScript接口后端无需任何适配层CI/CD流水线能自动校验每次模型更新是否破坏合约。我们曾用它捕获一次隐蔽的模型版本回滚事故——新模型在测试环境表现完美但上线后发现data.status_code字段从整数变成了字符串合约校验器在第一个请求就抛出异常避免了更大范围的故障。3. 实操实现从零搭建一个可运行的“02 组装 Pi”最小可行系统3.1 环境准备与依赖安装避开Python包版本地狱的实操技巧别急着pip install。我们踩过太多坑LangChain 0.1.x和0.2.x的Tool定义不兼容Pydantic V1和V2的BaseModel行为差异导致校验失效甚至某个Redis客户端版本在高并发下会静默丢弃EXPIRE命令。以下是经过生产环境验证的最小依赖清单requirements.txtlangchain-core0.1.49 langchain-community0.0.34 pydantic2.7.1 redis4.6.0 jsonschema4.21.1 jinja23.1.4注意必须锁定langchain-core和langchain-community的精确版本。LangChain官方文档常推荐langchain大包但它会自动拉取最新版子模块极易引发兼容问题。我们坚持“只装真正需要的”langchain-core提供基础Agent运行时langchain-community提供常用工具集成其余全部手动实现——这样虽然多写200行代码但换来的是线上三个月零因依赖升级导致的故障。安装时务必加--no-deps参数防止意外升级pip install --no-deps -r requirements.txt然后手动验证关键组件是否就位# 检查Pydantic是否为V2 python -c import pydantic; print(pydantic.VERSION) # 检查Redis连接假设本地运行 python -c import redis; r redis.Redis(); print(r.ping()) # 检查JSON Schema校验器 python -c from jsonschema import validate; print(OK)实测心得某次部署到客户现场服务器redis包默认安装了5.x版本导致EXPIRE命令返回None而非True/False上下文缓存永久不超时。解决方案不是降级而是加一层适配器# utils/redis_adapter.py import redis class SafeRedisClient: def __init__(self, *args, **kwargs): self.client redis.Redis(*args, **kwargs) def expire(self, name, time): # 兼容redis-py 4.x 和 5.x 的返回值差异 try: return bool(self.client.expire(name, time)) except Exception: return False这种“小封装大安心”的思路贯穿整个系统——不追求最新潮只求最稳定。3.2 Tool模块实战手写一个带完整契约的数据库查询工具我们以查询MySQL员工表为例展示如何将一个普通SQL函数升级为契约化Tool。第一步定义YAML契约文件tool/employee_db.yamlname: employee_db_query description: 执行安全的员工信息SQL查询仅支持SELECT自动过滤危险关键词 input_schema: type: object required: [sql] properties: sql: type: string maxLength: 500 description: 仅允许SELECT语句禁止INSERT/UPDATE/DELETE/DROP output_schema: type: object required: [rows, columns, query_time_ms] properties: rows: type: array description: 查询结果行数组 columns: type: array items: {type: string} description: 字段名列表 query_time_ms: {type: number, minimum: 0} error_handling: timeout: 5000 retry: 1 fallback: type: static value: {rows: [], columns: [name, email], query_time_ms: 0} observability: metrics_prefix: tool.employee_db log_fields: [sql, query_time_ms]第二步生成Pydantic模型并实现校验器# tool/employee_db.py from pydantic import BaseModel, field_validator from typing import List, Dict, Any import re class EmployeeDBInput(BaseModel): sql: str field_validator(sql) classmethod def validate_sql(cls, v): if not v.strip().upper().startswith(SELECT): raise ValueError(仅允许SELECT语句) if re.search(r(INSERT|UPDATE|DELETE|DROP|EXEC), v.upper()): raise ValueError(检测到危险SQL关键词) if len(v) 500: raise ValueError(SQL长度超限) return v class EmployeeDBOutput(BaseModel): rows: List[Dict[str, Any]] columns: List[str] query_time_ms: float field_validator(query_time_ms) classmethod def validate_time(cls, v): if v 0: raise ValueError(查询耗时不能为负) return v第三步实现带契约的工具执行器# tool/employee_db_executor.py import time import logging from pymysql import connect from tool.employee_db import EmployeeDBInput, EmployeeDBOutput logger logging.getLogger(__name__) class EmployeeDBExecutor: def __init__(self, host, user, password, database): self.config {host: host, user: user, password: password, database: database} def execute(self, input_data: dict) - dict: # 1. 输入校验 try: validated_input EmployeeDBInput(**input_data) except Exception as e: logger.error(f输入校验失败: {e}) return {rows: [], columns: [error], query_time_ms: 0} # 2. 执行查询带超时 start_time time.time() try: conn connect(**self.config, connect_timeout5) with conn.cursor() as cursor: cursor.execute(validated_input.sql) rows cursor.fetchall() columns [col[0] for col in cursor.description] query_time_ms (time.time() - start_time) * 1000 result { rows: [dict(zip(columns, row)) for row in rows], columns: columns, query_time_ms: query_time_ms } # 3. 输出校验 try: EmployeeDBOutput(**result) logger.info(f查询成功耗时{query_time_ms:.1f}ms) return result except Exception as e: logger.error(f输出校验失败: {e}) return {rows: [], columns: [error], query_time_ms: 0} except Exception as e: logger.error(f数据库执行异常: {e}) return {rows: [], columns: [error], query_time_ms: 0} finally: if conn in locals(): conn.close()第四步注册为LangChain Tool# agent/tool_registry.py from langchain_core.tools import StructuredTool from tool.employee_db_executor import EmployeeDBExecutor db_executor EmployeeDBExecutor( host127.0.0.1, userapp_user, passwordsecure_password, databasehr_db ) employee_db_tool StructuredTool.from_function( funcdb_executor.execute, nameemployee_db_query, description执行安全的员工信息SQL查询仅支持SELECT, args_schemaEmployeeDBInput, return_directFalse )这个工具现在具备了输入参数强校验、SQL注入防护、超时控制、结构化输出、失败降级、全链路日志。它不再是一个“能跑就行”的函数而是一个可监控、可审计、可替换的工程单元。3.3 Context模块实战三层缓存的代码实现与内存泄漏防护Session-level ContextRedis存储# context/session_manager.py import redis import json import time from typing import Dict, Any, Optional class SessionManager: def __init__(self, redis_client: redis.Redis): self.redis redis_client def get_session(self, session_id: str) - Dict[str, Any]: data self.redis.hgetall(fsess:{session_id}) if not data: return {} # Redis返回bytes需解码 return {k.decode(): json.loads(v.decode()) for k, v in data.items()} def set_session_field(self, session_id: str, key: str, value: Any, ttl_seconds: int 86400): # 使用HSET EXPIRE避免竞态条件 pipe self.redis.pipeline() pipe.hset(fsess:{session_id}, key, json.dumps(value)) pipe.expire(fsess:{session_id}, ttl_seconds) pipe.execute() def delete_session(self, session_id: str): self.redis.delete(fsess:{session_id}) # 初始化在应用启动时 redis_client redis.Redis(hostlocalhost, port6379, db0) session_manager SessionManager(redis_client)实操心得不要用SETEX存整个Session对象某次大促期间我们发现Redis内存暴涨排查发现是SETEX sess:abc {user_id:u123,prefs:{lang:zh}} 86400把整个JSON当字符串存了导致无法单独更新prefs.lang。改用Hash结构后内存占用下降62%且支持原子更新。Turn-level Context内存对象# context/turn_context.py from dataclasses import dataclass from datetime import datetime from typing import List, Dict, Any dataclass class TurnContext: turn_id: str session_id: str query: str tools_used: List[str] # 工具名列表 tool_results: Dict[str, Any] # 工具名 - 结果 summary: str # 本轮语义摘要 created_at: datetime def to_dict(self) - Dict[str, Any]: return { turn_id: self.turn_id, session_id: self.session_id, query: self.query, tools_used: self.tools_used, summary: self.summary, created_at: self.created_at.isoformat() } # 创建Turn上下文在每次请求入口 def create_turn_context(session_id: str, query: str) - TurnContext: return TurnContext( turn_idft-{int(time.time())}-{hash(query) % 10000}, session_idsession_id, queryquery, tools_used[], tool_results{}, summary, created_atdatetime.now() )Step-level Context局部变量# agent/step_executor.py from contextlib import contextmanager from typing import Dict, Any # Step上下文仅在with块内有效 contextmanager def step_context(step_name: str): step_data {step_name: step_name, start_time: time.time()} try: yield step_data finally: step_data[end_time] time.time() step_data[duration_ms] (step_data[end_time] - step_data[start_time]) * 1000 logger.info(fStep {step_name} completed in {step_data[duration_ms]:.1f}ms) # 使用示例 with step_context(db_query) as ctx: result db_executor.execute({sql: SELECT * FROM employees WHERE name张伟}) ctx[result] result # ctx只在此with块内可用自动清理三层Context的协同工作流如下请求到达create_turn_context()生成Turn对象Agent执行中每个工具调用包裹在step_context()内确保Step数据隔离工具返回后TurnContext.tool_results记录结果TurnContext.tools_used追加工具名Agent生成摘要后存入TurnContext.summary响应返回前TurnContext对象销毁其summary字段被提取并存入Session的last_summary字段供下轮参考。这种设计杜绝了内存泄漏——Turn对象生命周期与HTTP请求完全绑定Step对象随with块自动回收Session数据由Redis TTL自动清理。3.4 Output Contract实战从定义到校验的端到端实现第一步编写Output Contract Schemacontract/employee_answer.json{ $schema: https://json-schema.org/draft/2020-12/schema, type: object, required: [status, data, timestamp, request_id], properties: { status: { type: string, enum: [success, error, partial] }, data: { type: object, required: [email, department, manager], properties: { email: {type: string, format: email}, department: {type: string}, manager: { type: object, required: [name, emp_id], properties: { name: {type: string}, emp_id: {type: string, pattern: ^[A-Z]{2}\\d{4}$} } } } }, timestamp: {type: string, format: date-time}, request_id: {type: string, minLength: 10} } }第二步实现语义校验规则# contract/semantic_rules.py import re from datetime import datetime, timezone def validate_email_domain(data): 验证邮箱域名必须为公司域名 email data.get(data, {}).get(email, ) return in email and email.endswith(company.com) def validate_timestamp_freshness(data): 验证时间戳不能早于当前时间5分钟 try: ts datetime.fromisoformat(data[timestamp].replace(Z, 00:00)) now datetime.now(timezone.utc) return (now - ts).total_seconds() 300 except: return False def validate_manager_emp_id_format(data): 验证经理工号格式 manager data.get(data, {}).get(manager, {}) emp_id manager.get(emp_id, ) return bool(re.match(r^[A-Z]{2}\d{4}$, emp_id)) SEMANTIC_RULES { email_domain: validate_email_domain, timestamp_freshness: validate_timestamp_freshness, manager_emp_id_format: validate_manager_emp_id_format }第三步集成校验器到Agent响应流程# agent/output_validator.py import json import jsonschema from jsonschema import validate from jsonschema.exceptions import ValidationError from contract.semantic_rules import SEMANTIC_RULES class OutputContractValidator: def __init__(self, schema_path: str): with open(schema_path) as f: self.schema json.load(f) def validate(self, raw_output: str) - dict: # 阶段1JSON解析与结构校验 try: parsed json.loads(raw_output) except json.JSONDecodeError as e: raise ValueError(fJSON解析失败: {e}) try: validate(instanceparsed, schemaself.schema) except ValidationError as e: raise ValueError(f结构校验失败: {e.message}) # 阶段2语义校验 for rule_name, rule_func in SEMANTIC_RULES.items(): if not rule_func(parsed): raise ValueError(f语义校验失败: {rule_name}) # 阶段3一致性校验简化版检查data字段类型是否漂移 self._check_consistency(parsed) return parsed def _check_consistency(self, parsed: dict): # 实际项目中这里会查Redis或数据库的历史schema记录 # 简化版只检查data是否始终为object if not isinstance(parsed.get(data), dict): logger.warning(f检测到data字段类型漂移: 期望dict得到{type(parsed.get(data))}) # 在Agent响应后调用 validator OutputContractValidator(contract/employee_answer.json) try: clean_output validator.validate(llm_raw_response) return clean_output except ValueError as e: logger.error(f输出合约校验失败: {e}) # 触发fallback或重试逻辑 return {status: error, data: {}, timestamp: ..., request_id: ...}这个校验器现在成了Agent的“质量守门员”。它不关心LLM怎么思考只确保最终交付给下游系统的数据是结构确定、语义合规、时间可信的。上线后前端团队反馈TypeScript接口定义一次生成三个月未修改集成效率提升40%。4. 常见问题与避坑指南那些只有踩过才知道的细节4.1 Tool模块高频问题为什么我的工具总在重试后还是失败问题现象配置了retry: 2但日志显示工具连续3次调用都失败且第三次失败后没有触发fallback。根因分析LangChain的retry机制默认只捕获Exception而很多数据库驱动如pymysql在连接超时时抛出的是pymysql.err.OperationalError它继承自Exception但某些版本的LangChain会漏捕获。更隐蔽的是retry只针对工具执行函数本身不包括输入校验和输出校验。解决方案在工具执行器内部实现重试而非依赖LangChain# tool/employee_db_executor.py import time from functools import wraps def retry_on_failure(max_retries2, delay1): def decorator(func): wraps(func) def wrapper(*args, **kwargs): last_exception None for i in range(max_retries 1): try: return func(*args, **kwargs) except Exception as e: last_exception e if i max_retries: time.sleep(delay * (2 ** i)) # 指数退避 raise last_exception return wrapper return decorator class EmployeeDBExecutor: retry_on_failure(max_retries2, delay0.5) def execute(self, input_data: dict) - dict: # 原有逻辑不变 ...实操心得我们曾因此问题排查两天。LangChain的retry日志非常简略只显示“Retrying”不告诉你重试了几次、每次失败原因。在工具内部加retry后日志能清晰打印每次重试的详细错误定位速度提升5倍。4.2 Context模块高频问题Session缓存明明设置了TTL为什么还是不自动过期问题现象Redis中sess:abc的TTL显示-1永不过期导致僵尸会话堆积。根因分析EXPIRE命令只对key本身生效而我们用Hash结构存储SessionEXPIRE sess:abc 3600是对整个Hash key设TTL但如果代码中用了HSET sess:abc field valueRedis会重置TTL为-1。这是Redis的底层行为不是Bug。解决方案改用HSETEXPIREAT组合每次写入都刷新过期时间# context/session_manager.py def set_session_field(self, session_id: str, key: str, value: Any, ttl_seconds: int 86400): pipe self.redis.pipeline() pipe.hset(fsess:{session_id}, key, json.dumps(value)) # 计算绝对过期时间戳 expire_at int(time.time()) ttl_seconds pipe.expireat(fsess:{session_id}, expire_at) pipe.execute()注意EXPIREAT比EXPIRE更可靠因为它基于绝对时间戳不受Redis服务器时间漂移影响。我们在某台NTP未同步的测试机上验证过EXPIRE会因时间差导致提前过期EXPIREAT则稳定得多。4.3 Output Contract高频问题为什么JSON Schema校验通过了语义校验却总失败问题现象validate_against_schema()返回True但validate_email_domain()总返回False检查发现data.email字段是zhangcompany.com 末尾有空格。根因分析JSON Schema的string类型不校验首尾空格而业务规则要求邮箱必须精确匹配。这是Schema校验与业务校验的经典鸿沟。解决方案在语义校验前加一步“数据清洗”# contract/semantic_rules.py def validate_email_domain(data): email data.get(data, {}).get(email, ) email email.strip() # 关键清洗空格 return in email and email.endswith(company.com)更进一步我们在Output Contract Validator中加入通用清洗层def _normalize_data(self, data: dict) - dict: 对常见字段进行标准化清洗 normalized data.copy() if data in normalized and isinstance(normalized[data], dict): data_obj normalized[data] # 清洗邮箱 if email in data_obj: data_obj[email] str(data_obj[email]).strip() # 清洗姓名 if name in data_obj: data_obj[name] str(data_obj[name]).strip() return normalized def validate(self, raw_output: str) - dict: parsed json.loads(raw_output) parsed self._normalize_data(parsed) # 在校验前清洗 ...这个小技巧让我们避免了80%的“格式正确但业务错误”问题。记住Schema管结构清洗管格式语义管业务三者缺一不可。4.4 跨模块协同问题Tool返回的数据结构和Output Contract要求的不一致怎么平滑过渡问题现象老系统中employee_search工具返回{email: zhangcompany.com}但新Contract要求{data: {email: ...}}直接改Tool会导致所有调用方崩溃。解决方案引入“适配器模式”Adapter Pattern在Tool和Contract之间加一层转换# adapter/employee_search_adapter.py def adapt_employee_search_output(raw_result: dict) - dict: 将旧版工具输出适配为新版Contract要求 return { status: success, data: { email: raw_result.get(email, ), department: raw_result.get(department, ), manager: { name