DeepSeek-Agent-Harness-2026终极指南-第7章第31节-生产级日志与调用留痕-调用留痕五要素：每次模型调用的完整档案

DeepSeek Agent Harness 2026终极指南 - 第7章第31节 调用留痕五要素：每次模型调用的完整档案

上节日志系统就位了，但第 29 节的CallRecord还只是个草稿——塞在 client.py 里，字段随意，也没有持久化。这节把它升级为正式模块call_tracker.py：pydantic 强类型 + 六要素完整定义 + JSON 序列化落盘 + 按 call_id 唯一索引。这套设计是后续第 32 节 trace 落盘、第 33 节退出总结、第 34 节数据分析的共享基础。

本文导航

• CallRecord 为什么值得独立模块
• 六要素的工程含义
• CallRecord pydantic 升级版
• CallTracker：记录的汇总之地
• 与客户端集成
• 完整实录
• 小结

CallRecord 为什么值得独立模块

第 29 节我们在client.py里塞了个CallRecord，能用。但做到后面几节你会发现问题：

1. client.py 越来越胖。调用记录的数据模型、打印逻辑、存储逻辑全塞在客户端里，违反单一职责。客户端应该只管"发请求"，不管"怎么记"。

2. 其他模块也要用。第 33 节的退出总结需要读所有记录算 min/max/avg，第 34 节的数据分析要读记录做相关性分析。这些代码不应该 import client，而应该 import 一个纯粹的 call_tracker。

3. 字段设计需要正式化。第 29 节只是"先写出来用着"，没有 JSON Schema 生成、没有 timestamp 规范化、没有 call_id 唯一索引。这些后几节全要用。

所以独立出call_tracker.py——它是一个纯粹的记录仓库，不依赖 client、不依赖 logger，只关心"一条调用的完整档案长什么样"和"怎么存、怎么查"。

六要素的工程含义

"五条铁律"第三条对调用留痕的描述是：“远端 base_url 和端口、key 末尾 6 位、模型名、输入输出长度、起止时间年月日时分秒、总耗时”。拆开来，每个字段都有明确的工程用途：

为什么"年月日时分秒"这么细？因为"五条铁律"第七条规定退出时打印量化总结——你至少需要分钟级精度才能分清"同分钟内 10 次调用的耗时分布"。如果只记日期，一天一次，分析就废了。

CallRecord pydantic 升级版

把第 29 节的草稿升级为正式模块：

# deep_pilot/call_tracker.py —— 调用留痕核心模块 v0.2from__future__importannotationsimportjsonimportuuidfromdatetimeimportdatetimefrompathlibimportPathfromtypingimportClassVarfrompydanticimportBaseModel,Fieldfromdeep_pilot.configimportsettingsclassCallRecord(BaseModel):"""单次 API 调用的完整留痕档案——六要素 + 辅助字段"""# === 索引 ===call_id:str=Field(default_factory=lambda:uuid.uuid4().hex[:12],description="唯一调用标识，用 uuid4 前 12 位，碰撞概率可忽略")# === 六要素 ===base_url:str=Field(description="远端 API 地址，含端口")api_key_suffix:str=Field(description="密钥末尾 6 位，脱敏")model:str=Field(description="本次使用的模型名")input_length:int=Field(ge=0,description="输入 JSON 序列化后的字符数")output_length:int=Field(ge=0,description="输出 JSON 序列化后的字符数")prompt_tokens:int=Field(ge=0,description="API 返回的 prompt_tokens")completion_tokens:int=Field(ge=0,description="API 返回的 completion_tokens")start_time:datetime=Field(description="调用开始时间，精度到秒")end_time:datetime=Field(description="调用结束时间，精度到秒")elapsed_seconds:float=Field(ge=0,description="总耗时（秒）")finish_reason:str|None=Field(default=None,description="终止原因")# === 辅助 ===created_at:datetime=Field(default_factory=datetime.now,description="记录创建时间（服务端本地时间）")@propertydeftotal_tokens(self)->int:returnself.prompt_tokens+self.completion_tokensdefbrief(self)->str:"""一行摘要，用于控制台快速浏览"""return(f"[{self.call_id}]{self.model}| "f"in:{self.input_length}ch/{self.prompt_tokens}tk → "f"out:{self.output_length}ch/{self.completion_tokens}tk | "f"{self.elapsed_seconds:.1f}s |{self.finish_reason}")defto_log_line(self)->str:"""格式化为日志行（供 logger.info 使用）"""return(f"调用留痕 | call_id={self.call_id}| "f"端点={self.base_url}| key={self.api_key_suffix}| "f"模型={self.model}| "f"输入={self.input_length}字符/{self.prompt_tokens}token | "f"输出={self.output_length}字符/{self.completion_tokens}token | "f"开始={self.start_time:%Y-%m-%d%H:%M:%S}| "f"结束={self.end_time:%Y-%m-%d%H:%M:%S}| "f"耗时={self.elapsed_seconds:.2f}s | 终止={self.finish_reason}")classCallTracker:"""调用记录仓库——收集、存储、统计"""_instance:ClassVar[CallTracker|None]=Nonedef__new__(cls)->CallTracker:ifcls._instanceisNone:cls._instance=super().__new__(cls)cls._instance._records:list[CallRecord]=[]returncls._instance@propertydefrecords(self)->list[CallRecord]:returnself._records@propertydefcall_count(self)->int:returnlen(self._records)defadd(self,record:CallRecord)->None:self._records.append(record)defsummary(self)->dict:"""量化总结：min/max/avg 等统计指标"""ifnotself._records:return{"call_count":0,"message":"尚无调用记录"}prompt_tk=[r.prompt_tokensforrinself._records]comp_tk=[r.completion_tokensforrinself._records]total_tk=[r.total_tokensforrinself._records]elapsed=[r.elapsed_secondsforrinself._records]return{"call_count":len(self._records),"prompt_tokens":{"min":min(prompt_tk),"max":max(prompt_tk),"avg":round(sum(prompt_tk)/len(prompt_tk),1),"total":sum(prompt_tk),},"completion_tokens":{"min":min(comp_tk),"max":max(comp_tk),"avg":round(sum(comp_tk)/len(comp_tk),1),"total":sum(comp_tk),},"total_tokens":{"min":min(total_tk),"max":max(total_tk),"avg":round(sum(total_tk)/len(total_tk),1),"total":sum(total_tk),},"elapsed_seconds":{"min":round(min(elapsed),2),"max":round(max(elapsed),2),"avg":round(sum(elapsed)/len(elapsed),2),},"finish_reasons":_count_by(self._records,lambdar:r.finish_reasonor"unknown"),"models_used":_count_by(self._records,lambdar:r.model),}defdump_all(self,out_dir:Path|None=None)->Path:"""将所有记录序列化为 JSON 数组落盘"""target=(out_dirorsettings.log_dir)/"records"target.mkdir(parents=True,exist_ok=True)filepath=target/f"calls_{datetime.now():%Y%m%d_%H%M%S}.json"data=[r.model_dump(mode="json")forrinself._records]filepath.write_text(json.dumps(data,ensure_ascii=False,indent=2),encoding="utf-8")returnfilepathdef_count_by(items:list,key_fn)->dict:"""按某个属性分组计数，如 finish_reason 分布、模型分布"""result:dict={}foriteminitems:k=key_fn(item)result[k]=result.get(k,0)+1returndict(sorted(result.items(),key=lambdax:-x[1]))

CallTracker：记录的汇总之地

CallTracker用单例模式，全局一个仓库。各模块通过from deep_pilot.call_tracker import tracker拿到同一个实例。

关键设计：

call_id用 uuid4 前 12 位。12 位十六进制 = 48 位随机，碰撞概率约 1/2^48，在百万次调用量级下几乎为零。比自增 ID 的好处是不需要原子计数器，且删掉记录不影响 ID 序列。下节第 32 节的 trace 文件也用它做文件名前缀。

summary()用纯 Python 算 min/max/avg。不需要 pandas 或 numpy，记录量通常就在几千到几万条，列表推导 + 内置 max/min/sum 够快。_count_by是为分布统计（finish_reason 分布、模型分布）写的通用工具函数。

dump_all()做定期归档。虽然记录存内存够了，但程序跑了几天几夜后重启，记录丢了。第 33 节 atexit 会调dump_all()做退出前落盘。这里先写好方法，下下节直接调用。

to_log_line()和brief()两个格式化方法。to_log_line()是完整版，给 logger.info 用（六要素全部打出）。brief()是精简版，给需要快速浏览日志的场景。

与客户端集成

客户端从"自己管理记录"变为"丢给 CallTracker 管"：

# deep_pilot/client.py —— 改为用 call_trackerfromdeep_pilot.call_trackerimportCallRecord,CallTrackerfromdeep_pilot.loggerimportget_logger logger=get_logger(__name__)tracker=CallTracker()# 单例，全局一个classDeepSeekClient:def__init__(self,_settings:Any=settings)->None:# ... 原有代码 ...pass# 删掉 self._call_records 和 self._call_count 等# 所有记录和统计全部委托给 trackerdef_build_response(self,completion,messages,start:datetime)->ChatResponse:# ... 前面一样，拿到 choice, msg, end, usage ...# 构建 CallRecord 并存入 trackerrecord=CallRecord(base_url=self._settings.deepseek_base_url,api_key_suffix=self._settings.api_key_masked,model=completion.model,input_length=len(json.dumps(messages,ensure_ascii=False)),output_length=len(json.dumps(msg.model_dump()ifhasattr(msg,"model_dump")else{"content":msg.content},ensure_ascii=False,)),prompt_tokens=usage.prompt_tokensifusageelse0,completion_tokens=usage.completion_tokensifusageelse0,start_time=start,end_time=end,elapsed_seconds=(end-start).total_seconds(),finish_reason=choice.finish_reason,)tracker.add(record)# 统一的记录仓库logger.info(record.to_log_line())# 统一的日志格式returnChatResponse(...)

改动很小——把原来self._call_records.append(record)换成tracker.add(record)，把原来手拼 log string 换成record.to_log_line()。客户端瘦了，逻辑更干净。

完整实录

把call_tracker.py写好，跑一遍验证：

cddeep-pilot uv run python-c" from deep_pilot.client import client from deep_pilot.call_tracker import tracker # 三次调用 client.chat([{'role': 'user', 'content': '你好'}]) client.chat([{'role': 'user', 'content': '解释一下 Python 装饰器'}]) client.chat([{'role': 'user', 'content': '写一个排序算法'}]) # 看 summary import json print(json.dumps(tracker.summary(), indent=2, ensure_ascii=False)) print(f'记录总数: {tracker.call_count}') for r in tracker.records: print(f' {r.brief()}') "

控制台输出：

2026-09-12 10:00:01 | INFO | deep_pilot.client | 调用留痕 | call_id=a1b2c3d4e5f6 | 端点=https://api.deepseek.com/v1 | key=sk-...abc123 | 模型=deepseek-flash | 输入=48字符/8token | 输出=38字符/6token | 开始=2026-09-12 10:00:01 | 结束=2026-09-12 10:00:02 | 耗时=1.15s | 终止=stop 2026-09-12 10:00:02 | INFO | deep_pilot.client | 调用留痕 | call_id=b2c3d4e5f6a1 | 端点=https://api.deepseek.com/v1 | key=sk-...abc123 | 模型=deepseek-flash | 输入=62字符/15token | 输出=520字符/142token | 开始=2026-09-12 10:00:02 | 结束=2026-09-12 10:00:06 | 耗时=3.82s | 终止=stop 2026-09-12 10:00:06 | INFO | deep_pilot.client | 调用留痕 | call_id=c3d4e5f6a1b2 | 端点=https://api.deepseek.com/v1 | key=sk-...abc123 | 模型=deepseek-flash | 输入=50字符/10token | 输出=892字符/234token | 开始=2026-09-12 10:00:06 | 结束=2026-09-12 10:00:11 | 耗时=5.37s | 终止=stop { "call_count": 3, "prompt_tokens": {"min": 8, "max": 15, "avg": 11.0, "total": 33}, "completion_tokens": {"min": 6, "max": 234, "avg": 127.3, "total": 382}, "total_tokens": {"min": 14, "max": 244, "avg": 138.3, "total": 415}, "elapsed_seconds": {"min": 1.15, "max": 5.37, "avg": 3.45}, "finish_reasons": {"stop": 3}, "models_used": {"deepseek-flash": 3} } 记录总数: 3 [a1b2c3d4e5f6] deepseek-flash | in:48ch/8tk → out:38ch/6tk | 1.1s | stop [b2c3d4e5f6a1] deepseek-flash | in:62ch/15tk → out:520ch/142tk | 3.8s | stop [c3d4e5f6a1b2] deepseek-flash | in:50ch/10tk → out:892ch/234tk | 5.4s | stop

三条记录，三个 call_id。summary()输出了完整的 min/max/avg 分布——输出越长耗时越多（6 token → 1.1s vs 234 token → 5.4s），合理。finish_reasons和models_used的分布统计也为后续多模型 A/B 测试留了接口。

小结

1. CallRecord 独立模块 = 单一职责。客户端只管发请求，记录管理全在 call_tracker，各取所需不交叉。
2. 六要素设计有工程意图：base_url 定端点、key_suffix 定来源、model 定基准、length 算成本、时间算吞吐、耗时算性能。
3. call_id 用 uuid4 前 12 位：唯一索引，不依赖自增计数器，trace 文件名也用它做前缀。
4. summary()纯 Python 算 min/max/avg：不需要额外依赖，记录量<10万时性能足够。
5. tracker单例全局共享：客户端写入记录，第 33 节 atexit 和 第 34 节分析都读同一个实例。

下节预告

CallRecord 记录了六要素，但有个致命盲区——原始输入（messages）和原始输出（model_dump）没存。下次调试想复现问题时，你只能看到"输入 48 字符"，看不到那 48 字符到底是什么。下一节做trace 落盘：每次调用的完整 messages JSON 和 response JSON 写入logs/trace/目录，按 call_id 命名，支持事后精确复现。这才是"追溯"的完整闭环。

如果觉得本文对你有帮助，欢迎点赞、收藏、关注三连！ 本系列持续更新中，关注不迷路~