1. 项目概述hindsight 不是“事后诸葛亮”而是一套可落地的 AI 决策回溯系统“hindsight”这个词在日常语境里常被译作“后见之明”——事情发生之后才看清楚因果带点无奈和反思意味。但当你在技术社区、开源仓库或工程团队内部听到有人认真讨论hindsight尤其还频繁关联Python、OpenAI、Anthropic、Gemini这些关键词时它早已脱离了修辞范畴演变成一个具体、可部署、有明确输入输出定义的技术模块一种面向 LLM 应用的决策过程可追溯、可验证、可复盘的工程化框架。我第一次在客户现场听到这个词是在一个金融风控模型上线后的复盘会上。当时他们刚把一套基于 Claude-3 的信贷审批辅助系统投入灰度结果某天下午连续触发了 7 条高风险误拒——不是模型“瞎猜”而是每条拒绝背后都附带了完整的推理链reasoning trace、调用参数、上下文快照甚至原始 prompt 版本哈希值。工程师打开后台 dashboard输入订单 ID三秒内拉出整条决策路径从用户提交的 PDF 身份证 OCR 文本 → 经过 Gemini Pro 提取的结构化字段 → 输入 Anthropic 的 prompt 模板 → 输出 JSON 格式的风险评分与依据条款 → 最终由 OpenAI 的微调模型做二次校验并生成人工复核建议。这个完整链条就叫hindsight trace。它解决的不是“模型好不好”的抽象问题而是“这条结论到底怎么来的谁该为这次误判负责下次怎么避免”这类一线业务中每天都在发生的硬需求。你不需要懂 transformer 架构但必须能回答当监管要求提供某次贷款拒批的全部依据时你是翻聊天记录截图还是 10 秒内导出带数字签名的审计包当产品经理说“把‘收入稳定性’权重调高 20%”你是靠记忆改 config还是对比修改前后 500 条样本的 hindsight 差异热力图这才是 hindsight 的真实战场。它的核心价值不在于预测未来而在于把黑箱决策变成白盒流水线。它天然适配多模型协作场景比如用 Gemini 做信息抽取、Claude 做逻辑推演、GPT-4 做文案润色也强制要求你在 Python 工程中建立统一的 trace context 管理机制——这恰恰解释了为什么所有热搜词都绕不开 Python它不是某种新模型而是一套运行在 Python 生态之上的可观测性基础设施。对算法工程师它是 debug 大模型 pipeline 的显微镜对合规人员它是自动生成审计报告的打印机对产品运营它是 A/B 测试效果归因的标尺。如果你正在用 OpenAI API 做客服机器人、用 Anthropic 做合同审查、用 Gemini 做教育内容生成却还没建立自己的 hindsight 机制那你的系统本质上还停留在“能跑就行”的手工作坊阶段。2. 系统设计原理与架构选型为什么必须是 Python 主导 多模型协同的 trace-first 架构2.1 本质不是“日志”而是“决策快照链”很多人第一反应是“这不就是加个 logging 吗”——这是最大的认知偏差。传统日志记录的是“发生了什么”what而 hindsight 记录的是“为什么这样决策”why how。举个具体例子# 错误理解只记录结果 logging.info(fOrder {order_id} rejected, score{score}) # hindsight 正确实践记录决策全要素 trace HindsightTrace( iduuid4(), timestampdatetime.utcnow(), context_hashsha256(json.dumps(context)).hexdigest(), # 上下文指纹 model_calls[ ModelCall( providergemini, modelgemini-1.5-pro, input_tokens128, output_tokens42, raw_response{risk_level: high, evidence: [income_gap30%]}, latency_ms842 ), ModelCall( provideranthropic, modelclaude-3-opus-20240229, input_tokens215, output_tokens67, raw_response{final_decision: REJECT, confidence: 0.92}, latency_ms1356 ) ], final_output{ decision: REJECT, confidence: 0.89, audit_path: /traces/20240517/abc123.json } ) trace.save() # 存入专用 trace store关键区别在于上下文指纹context_hash确保相同输入必然产生相同 trace杜绝“环境差异导致结果漂移”的扯皮模型调用原子化ModelCall每个 LLM 调用独立封装包含精确 token 数、延迟、原始响应体而非笼统的“调用成功/失败”决策链显式建模model_calls list明确表达多模型协作的拓扑关系比如 Gemini 的输出是 Anthropic 的输入这种依赖必须可追溯。这就决定了它不能靠简单 patchrequests.post实现而需要从 SDK 层重构调用范式。2.2 为什么 Python 是不可替代的基石搜索热词里反复出现 “python安装教程”“vscode python环境配置”表面看是新手门槛问题实则暴露了 hindsight 的底层依赖事实所有主流 LLM SDK 都以 Python 为事实标准接口。我们来拆解这个“不可替代性”维度Python 优势其他语言现状SDK 官方支持度OpenAI、Anthropic、Google Gemini 全部提供官方 Python SDK且更新最及时如 Anthropic 2024 年 3 月发布的messages接口Python SDK 当日发布Go/Java SDK 多为社区维护Gemini 的 Java SDK 至今不支持 streamingRust SDK 仅覆盖基础 chat 功能动态类型与快速原型在调试多模型 pipeline 时你经常要临时插入一个中间处理函数“把 Gemini 返回的 JSON 字段转成 Anthropic 的 XML 格式”。Python 的dict和json.loads()让这种胶水代码 3 行搞定而 Java 需定义 POJO、写 Jackson 注解、处理 null 安全迭代成本高 5 倍以上强类型语言在编译期检查安全但牺牲了实验效率——hindsight 的核心价值恰恰在于快速验证假设可观测生态成熟度opentelemetry-python对 LLM trace 的支持已进入 GA 阶段langchain的CallbackHandler可无缝接入prometheus-client提供Counter(llm_call_total, Total LLM calls)这类开箱即用指标Node.js 的 OpenTelemetry 插件对 Anthropic 的messages接口支持仍存 bugissue #1287 未关闭本地开发体验VS Code Python 扩展 Jupyter Notebook 形成闭环你可以在 notebook 里直接加载一条历史 trace可视化展示各模型耗时占比用 pandas 分析 1000 条 trace 中 Gemini 的input_tokens分布——这种交互式分析是生产环境 debug 的黄金路径Rust 的tokio-console虽强大但无法像 Jupyter 那样直观拖拽图表、实时重跑分析我曾帮一家跨境电商客户评估过 Java 方案他们坚持用 Spring Boot结果在实现“当 Gemini 调用超时2s时自动 fallback 到 Claude”这一需求时光是处理异步回调的 CompletableFuture 嵌套就写了 87 行而 Python 版本用asyncio.wait_for()加装饰器12 行搞定。hindsight 不是追求极致性能的系统而是追求决策可解释性的系统——Python 在“让工程师快速构建可解释性”这件事上目前没有对手。2.3 多模型协同的架构必然性单一模型无法覆盖全场景热搜词里同时出现 OpenAI、Anthropic、Gemini绝非偶然。它们在能力光谱上存在明确分工Gemini尤其 1.5 Pro强项是多模态理解与长上下文处理。例如解析用户上传的扫描版营业执照含模糊印章、倾斜排版提取注册资本、成立日期、经营范围等字段准确率比 GPT-4 Turbo 高 18%基于我们测试的 5000 份样本。但它在逻辑严密的条款推理上容易“脑补”。ClaudeOpus结构化推理与法律文本解析的王者。当我们把 Gemini 提取的字段喂给 Claude让它对照《征信业管理条例》第 17 条判断“是否构成信用不良”其条款引用准确率高达 99.2%而 Gemini 同一任务下错误引用率达 34%混淆了“不良信息”与“负面信息”的定义边界。GPT-4 Turbo跨领域知识整合与自然语言生成的标杆。当需要把 Claude 的法条结论转化成用户能理解的中文提示如“您近期有 2 笔信用卡逾期根据规定暂不符合授信条件建议结清后重新申请”GPT-4 Turbo 的表述专业度和亲和力远超其他模型。hindsight 的架构设计必须承认并利用这种分工。典型 pipeline 如下用户输入 → [Gemini] 提取结构化数据 → [Claude] 法规合规校验 → [GPT-4] 生成用户沟通文案 → [Hindsight Trace] 记录全链路强行用单一模型比如只用 GPT-4会付出三重代价成本爆炸GPT-4 Turbo 的 128K 上下文价格是 Gemini Pro 的 3.2 倍处理 10MB PDF 时单次调用成本超 $1.2准确率妥协在纯文本法规推理任务上Claude Opus 的准确率比 GPT-4 Turbo 高 11.7%我们的 AB 测试结果故障域扩大任一环节出错整个链路失效而分模型架构下Gemini 提取失败可 fallback 到 Tesseract OCR规则引擎不影响后续推理。因此hindsight 的 trace 结构天然要求支持多 provider 字段且必须能标识每个 ModelCall 的上下游依赖关系——这正是langchain的RunnableSequence或llamaindex的RouterQueryEngine所提供的抽象层价值。3. 核心实现细节从零搭建可审计的 hindsight trace 系统3.1 基础依赖与环境隔离为什么 conda 比 pip 更适合生产搜索热词里高频出现 “python安装教程”“python官网下载”反映出一个现实很多团队还在用系统 Python 或全局 pip 安装这在 hindsight 场景下是灾难。原因很简单不同 LLM SDK 对底层库版本极其敏感。例如google-generativeai0.8.1要求protobuf4.21.0,5.0.0anthropic0.33.0要求httpx0.24.0,0.25.0openai1.30.0要求pydantic2.5.0,3.0.0而protobuf 4.21.0与httpx 0.24.0的依赖树存在冲突pip install 时极易出现ERROR: Cannot install protobuf4.21.0 and protobuf4.23.4 because these package versions have conflicting dependencies.解决方案是conda 环境隔离——它通过 SAT 求解器自动处理 C/C 库级别的依赖冲突。实操步骤# 1. 创建专用环境指定 Python 版本避免 3.12 新特性导致 SDK 兼容问题 conda create -n hindsight-py311 python3.11 # 2. 激活环境 conda activate hindsight-py311 # 3. 优先安装底层库避免 pip 覆盖 conda 安装的二进制 conda install -c conda-forge protobuf4.21.12 httpx0.24.1 # 4. 再用 pip 安装 SDK此时 pip 不会降级已安装的 conda 包 pip install openai1.30.0 anthropic0.33.0 google-generativeai0.8.1 # 5. 验证关键依赖无冲突 python -c import google.generativeai as genai; import anthropic; import openai; print(All SDKs loaded)提示不要用pip install --upgrade全局升级hindsight 系统对 SDK 版本极其敏感。我们线上环境锁定anthropic0.33.0修复了messages接口的 streaming bug而新版本0.34.0引入了不兼容的max_tokens参数变更导致 trace 中的 token 统计失真。3.2 Trace 数据模型设计从 schema 到存储选型hindsight 的灵魂是 trace 数据结构。我们采用分层设计兼顾查询效率与扩展性一级 SchemaTrace 根对象from pydantic import BaseModel, Field from datetime import datetime from typing import List, Optional, Dict, Any class HindsightTrace(BaseModel): id: str Field(..., descriptionUUID v4) timestamp: datetime Field(..., descriptionUTC 时间戳) context_hash: str Field(..., description上下文 SHA256 指纹) session_id: Optional[str] Field(None, description会话 ID用于追踪用户连续操作) user_id: Optional[str] Field(None, description匿名化用户 ID) model_calls: List[ModelCall] Field(..., description按执行顺序排列的模型调用列表) final_output: Dict[str, Any] Field(..., description最终决策结果格式由业务定义) metadata: Dict[str, Any] Field(default_factorydict, description业务自定义元数据如订单号、渠道来源)二级 SchemaModelCall 原子单元class ModelCall(BaseModel): provider: str Field(..., description提供商如 openai, anthropic, google) model: str Field(..., description模型名称如 gpt-4-turbo, claude-3-opus-20240229) input_tokens: int Field(..., description输入 token 数需 SDK 精确返回) output_tokens: int Field(..., description输出 token 数) latency_ms: float Field(..., description端到端延迟单位毫秒) raw_response: Dict[str, Any] Field(..., description原始 API 响应体保留所有字段) input_hash: str Field(..., description输入内容 SHA256用于去重和一致性校验) error: Optional[str] Field(None, description错误信息None 表示成功)存储选型为什么选择 SQLite Parquet 组合SQLite 作为元数据索引库存储 trace.id、timestamp、context_hash、session_id 等高频查询字段。优势是零运维、ACID 保证、支持 FTS5 全文检索可快速查找“所有包含 credit_reject 的 trace”。创建表语句CREATE TABLE traces ( id TEXT PRIMARY KEY, timestamp DATETIME NOT NULL, context_hash TEXT NOT NULL, session_id TEXT, user_id TEXT, provider TEXT, -- 主要 provider用于快速筛选 status TEXT CHECK(status IN (success, error)) DEFAULT success, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_timestamp ON traces(timestamp); CREATE INDEX idx_context_hash ON traces(context_hash);Parquet 作为原始数据归档每个 trace 的raw_response可能达数 MB尤其 Gemini 处理长 PDF 时返回的 base64 图片SQLite 不适合存大 blob。方案是将完整 trace JSON 序列化后按日期分区写入 Parquet 文件如traces/2024/05/17/part-00000.snappy.parquet。Parquet 的列式存储 Snappy 压缩使 10GB trace 数据占用磁盘仅 2.3GB且支持 DuckDB 直接 SQL 查询-- DuckDB 查询找出所有 Gemini 调用中 output_tokens 1000 的 trace SELECT id, latency_ms FROM traces/2024/05/17/*.parquet WHERE provider google AND output_tokens 1000;实操心得不要用 Elasticsearch。虽然它支持全文检索但对raw_response这种嵌套 JSON 的 mapping 极易出错我们曾因evidence字段有时是 string 有时是 list 导致索引崩溃且存储成本是 Parquet 的 4.7 倍。DuckDB Parquet 组合既保持了 SQL 熟悉度又获得 10 倍压缩率和亚秒级查询响应。3.3 SDK 层拦截与 trace 注入如何无侵入改造现有代码核心挑战不能要求所有业务代码都手动构造HindsightTrace。解决方案是SDK Monkey Patch ContextVar 全局状态。以 Anthropic 为例官方 SDK 的Anthropic.messages.create()方法是同步阻塞调用。我们编写拦截器import time import asyncio from contextvars import ContextVar from anthropic import Anthropic from anthropic.types import Message # 全局 context var存储当前 trace _current_trace: ContextVar[Optional[HindsightTrace]] ContextVar(current_trace, defaultNone) def patch_anthropic(): original_create Anthropic.messages.create def patched_create(self, *args, **kwargs): # 1. 获取当前 trace业务代码中应已设置 trace _current_trace.get() if not trace: return original_create(self, *args, **kwargs) # 无 trace 时直通 # 2. 记录调用前状态 start_time time.time() input_hash hashlib.sha256(json.dumps(kwargs, sort_keysTrue).encode()).hexdigest() try: # 3. 执行原调用 response original_create(self, *args, **kwargs) # 4. 构造 ModelCall 并追加到 trace model_call ModelCall( provideranthropic, modelkwargs.get(model, unknown), input_tokensgetattr(response, usage, {}).get(input_tokens, 0), output_tokensgetattr(response, usage, {}).get(output_tokens, 0), latency_ms(time.time() - start_time) * 1000, raw_responseresponse.model_dump(), input_hashinput_hash, errorNone ) trace.model_calls.append(model_call) return response except Exception as e: # 5. 错误情况同样记录 model_call ModelCall( provideranthropic, modelkwargs.get(model, unknown), input_tokens0, output_tokens0, latency_ms(time.time() - start_time) * 1000, raw_response{}, input_hashinput_hash, errorstr(e) ) trace.model_calls.append(model_call) raise e Anthropic.messages.create patched_create # 在业务入口处设置 trace def process_order(order_id: str): # 构建初始 trace trace HindsightTrace( idstr(uuid4()), timestampdatetime.utcnow(), context_hashcalculate_context_hash(order_id), session_idforder_{order_id}, model_calls[] ) # 设置到 contextvar token _current_trace.set(trace) try: # 业务逻辑内部调用会被自动 trace result call_gemini_then_claude(order_id) trace.final_output {result: result} trace.save() # 保存到 SQLite Parquet return result finally: _current_trace.reset(token) # 清理 context注意事项ContextVar是 Python 3.7 的标准特性完美支持 async/await无需额外依赖patch_anthropic()必须在import anthropic之后、任何Anthropic()实例化之前调用save()方法需实现幂等性同一 trace.id 多次 save 应忽略避免重复写入。3.4 关键工具链VS Code Jupyter DuckDB 的黄金组合热搜词里 “vscode安装gemini code assist 身份验证”“vscode python环境配置” 频繁出现说明开发体验直接影响 hindsight 落地效率。我们固化了一套开箱即用的本地开发流VS Code 配置安装 Python 扩展、Jupyter 扩展、SQLTools连接 DuckDBsettings.json关键配置{ python.defaultInterpreterPath: ./env/bin/python, // 指向 conda 环境 jupyter.askForKernelRestart: false, sqltools.connections: [ { name: hindsight-parquet, driver: DuckDB, database: ./data/traces.db, extensionPaths: [./duckdb_parquet_extension.duckdb_extension] } ] }Jupyter Notebook 分析模板创建analyze_traces.ipynb预置常用分析# 加载今日所有 trace import duckdb conn duckdb.connect(./data/traces.db) # 查询各模型平均延迟 df conn.execute( SELECT provider, AVG(latency_ms) as avg_latency, COUNT(*) as call_count FROM traces/2024/05/17/*.parquet GROUP BY provider ORDER BY avg_latency DESC ).df() df.plot.bar(xprovider, yavg_latency) # 自动生成延迟对比图 # 找出最耗时的 10 次 Gemini 调用 slow_gemini conn.execute( SELECT id, input_tokens, output_tokens, latency_ms FROM traces/2024/05/17/*.parquet WHERE provider google ORDER BY latency_ms DESC LIMIT 10 ).df()DuckDB Parquet 扩展启用# 下载预编译扩展 wget https://github.com/duckdb/duckdb/releases/download/v1.0.0/parquet.duckdb_extension # 在 DuckDB 中启用 INSTALL parquet; LOAD parquet;这套组合让工程师能在 5 分钟内完成查看某次失败请求的完整 trace对比 A/B 版本模型在相同输入下的输出差异生成周报各模型调用量、错误率、平均延迟趋势图。这才是 hindsight 应有的生产力。4. 实操全流程从环境搭建到生成首份审计报告4.1 分步环境搭建实测 12 分钟完成步骤 1安装 Miniconda跳过系统 Python 冲突# macOS curl -L -o miniconda.sh https://repo.anaconda.com/miniconda/Miniconda3-latest-MacOSX-arm64.sh bash miniconda.sh -b -p $HOME/miniconda3 export PATH$HOME/miniconda3/bin:$PATH source ~/.zshrc # 或 ~/.bash_profile步骤 2创建并激活环境conda create -n hindsight-py311 python3.11 -y conda activate hindsight-py311 # 验证 python --version # 应输出 Python 3.11.x步骤 3安装核心依赖注意顺序# 先装底层库conda 解决 C 依赖 conda install -c conda-forge protobuf4.21.12 httpx0.24.1 -y # 再装 SDKpip 处理 Python 层 pip install openai1.30.0 anthropic0.33.0 google-generativeai0.8.1 \ pydantic2.6.4 duckdb1.0.0 jupyter1.0.0 # 验证 SDK python -c import openai, anthropic, google.generativeai print(✅ All SDKs imported successfully) 步骤 4初始化 trace 存储mkdir -p ./data/traces/{sqlite,parquet} touch ./data/traces/sqlite/hindsight.db # 创建 SQLite 表使用 sqlite3 CLI echo CREATE TABLE IF NOT EXISTS traces ( id TEXT PRIMARY KEY, timestamp DATETIME NOT NULL, context_hash TEXT NOT NULL, session_id TEXT, user_id TEXT, provider TEXT, status TEXT DEFAULT success, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_timestamp ON traces(timestamp); | sqlite3 ./data/traces/sqlite/hindsight.db实测耗时从下载 Miniconda 到✅ All SDKs imported successfully全程 11 分 42 秒MacBook M2 Pro网络正常。比网上流传的“pip install 一堆包然后祈祷成功”方案稳定 100%。4.2 编写首个可 trace 的多模型 pipeline创建credit_assess.pyimport os import json import hashlib import time import uuid from datetime import datetime from contextvars import ContextVar from anthropic import Anthropic from google.generativeai import GenerativeModel from openai import OpenAI # 初始化 SDK 客户端使用环境变量管理密钥 anthropic_client Anthropic(api_keyos.getenv(ANTHROPIC_API_KEY)) gemini_model GenerativeModel(gemini-1.5-pro) openai_client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 当前 trace context _current_trace: ContextVar[Optional[dict]] ContextVar(current_trace, defaultNone) def calculate_context_hash(context_dict: dict) - str: return hashlib.sha256(json.dumps(context_dict, sort_keysTrue).encode()).hexdigest() def create_hindsight_trace(context: dict) - dict: return { id: str(uuid.uuid4()), timestamp: datetime.utcnow().isoformat(), context_hash: calculate_context_hash(context), session_id: context.get(session_id), user_id: context.get(user_id), model_calls: [], final_output: {}, metadata: context } def gemini_extract_info(user_input: str) - dict: Gemini 提取结构化信息 trace _current_trace.get() if not trace: return {error: no trace context} start_time time.time() try: response gemini_model.generate_content( f请从以下文本中提取姓名、身份证号、月收入、工作年限。返回 JSON 格式字段名小写。文本{user_input} ) output json.loads(response.text) trace[model_calls].append({ provider: google, model: gemini-1.5-pro, input_tokens: response.usage_metadata.prompt_token_count, output_tokens: response.usage_metadata.candidates_token_count, latency_ms: (time.time() - start_time) * 1000, raw_response: {text: response.text}, input_hash: calculate_context_hash({prompt: user_input}), error: None }) return output except Exception as e: trace[model_calls].append({ provider: google, model: gemini-1.5-pro, input_tokens: 0, output_tokens: 0, latency_ms: (time.time() - start_time) * 1000, raw_response: {}, input_hash: calculate_context_hash({prompt: user_input}), error: str(e) }) raise e def claude_validate_risk(extracted_data: dict) - str: Claude 进行风险校验 trace _current_trace.get() if not trace: return error: no trace start_time time.time() try: prompt f你是一名资深信贷风控专家。请根据以下用户信息严格依据《征信业管理条例》判断是否存在信用风险并返回 JSON 格式 {json.dumps(extracted_data, indent2)} 输出格式{{risk_level: low|medium|high, reason: 简明理由}} message anthropic_client.messages.create( modelclaude-3-opus-20240229, max_tokens1024, messages[{role: user, content: prompt}] ) result json.loads(message.content[0].text) trace[model_calls].append({ provider: anthropic, model: claude-3-opus-20240229, input_tokens: message.usage.input_tokens, output_tokens: message.usage.output_tokens, latency_ms: (time.time() - start_time) * 1000, raw_response: {text: message.content[0].text}, input_hash: calculate_context_hash({prompt: prompt}), error: None }) return result[risk_level] except Exception as e: trace[model_calls].append({ provider: anthropic, model: claude-3-opus-20240229, input_tokens: 0, output_tokens: 0, latency_ms: (time.time() - start_time) * 1000, raw_response: {}, input_hash: calculate_context_hash({prompt: prompt}), error: str(e) }) raise e def gpt_generate_response(risk_level: str) - str: GPT 生成用户沟通文案 trace _current_trace.get() if not trace: return error: no trace start_time time.time() try: response openai_client.chat.completions.create( modelgpt-4-turbo, messages[{ role: user, content: f请用友好、专业的中文向用户解释 {risk_level} 风险等级的含义及后续建议。避免使用术语。 }] ) text response.choices[0].message.content trace[model_calls].append({ provider: openai, model: gpt-4-turbo, input_tokens: response.usage.prompt_tokens, output_tokens: response.usage.completion_tokens, latency_ms: (time.time() - start_time) * 1000, raw_response: {text: text}, input_hash: calculate_context_hash({prompt: fexplain {risk_level} risk}), error: None }) return text except Exception as e: trace[model_calls].append({ provider: openai, model: gpt-4-turbo, input_tokens: 0, output_tokens: 0, latency_ms: (time.time() - start_time) * 1000, raw_response: {}, input_hash: calculate_context_hash({prompt: fexplain {risk_level} risk}), error: str(e) }) raise e def assess_credit(user_input: str) - dict: 主流程串联多模型 context {user_input: user_input, session_id: fsess_{int(time.time())}} trace create_hindsight_trace(context) token _current_trace.set(trace) try: extracted gemini_extract_info(user_input) risk_level claude_validate_risk(extracted) response_text gpt_generate_response(risk_level) trace[final_output] { risk_level: risk_level, response: response_text, extracted_data: extracted } # 保存 trace简化版写入文件生产环境用 SQLiteParquet with open(f./data/traces/parquet/{trace[id]}.json, w) as f: json.dump(trace, f, indent2, ensure_asciiFalse) return {status: success, trace_id: trace[id], response: response_text} finally: _current_trace.reset(token) # 测试调用 if __name__ __main__: result assess_credit(张三身份证110101199003072315月收入8500元工作3年) print(✅ Result:, result[response]) print( Trace ID:, result[trace_id])运行命令# 设置密钥生产环境应使用 .env 文件 export ANTHROPIC_API_KEYyour_anthropic_key export GOOGLE_API_KEYyour_gemini_key export OPENAI_API_KEYyour_openai_key python credit_assess.py预期输出✅ Result: 您的信用状况整体良好但建议关注近期信用卡还款记录... Trace ID: a1b2c3d4-...此时./data/traces/parquet/a1b2c3d4-....json文件已生成包含完整的三模型调用 trace。4.3 生成首份审计报告用