137 lines
5.6 KiB
Python
137 lines
5.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""对话日志模块。
|
||
|
||
设计目标：每个总排号一个日志文件，纯文本格式，完整记录"这次调用到底发生了
|
||
什么"——system prompt、few-shot、实际发给模型的用户消息、模型每一次原始
|
||
回复（含被格式校验拒绝、触发重试的那些），以及最终的解析结果。成功失败都写，
|
||
不因为调用失败就少记东西；恰恰是失败的调用最需要完整日志去排查。
|
||
|
||
文件命名: {log_dir}/{总排号}_{时间戳}.log
|
||
同一个总排号如果被反复处理（比如手动重跑），不会覆盖旧日志，每次生成新文件，
|
||
方便对比"改了提示词前后，同一条记录的判断有没有变化"。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from llm_client import LLMCallResult
|
||
from parser import ParsedResult
|
||
|
||
# 总排号可能含有对文件名不友好的字符（如斜杠），做一次保守清理
|
||
_UNSAFE_FILENAME_CHARS = re.compile(r'[\\/:*?"<>|]')
|
||
|
||
|
||
def _safe_filename_part(zong_pai_hao: str) -> str:
|
||
return _UNSAFE_FILENAME_CHARS.sub("_", zong_pai_hao)
|
||
|
||
|
||
class OrderLogger:
|
||
"""负责单个总排号在一次分类过程中的完整日志记录。
|
||
|
||
典型用法（在 classifier.py 里）：
|
||
logger = OrderLogger(log_dir, zong_pai_hao, mode)
|
||
logger.log_param_text(param_text)
|
||
# ... 每次调用LLM后:
|
||
logger.log_llm_attempt(attempt_no, call_result, parsed_or_none, format_error_or_none)
|
||
# ... 最终:
|
||
logger.log_final_result(status, parsed_result_or_none)
|
||
logger.close()
|
||
"""
|
||
|
||
def __init__(self, log_dir: str | Path, zong_pai_hao: str, mode: str):
|
||
self._log_dir = Path(log_dir)
|
||
self._log_dir.mkdir(parents=True, exist_ok=True)
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
|
||
filename = f"{_safe_filename_part(zong_pai_hao)}_{timestamp}.log"
|
||
self._path = self._log_dir / filename
|
||
self._lines: list[str] = []
|
||
|
||
self._write_header(zong_pai_hao, mode)
|
||
|
||
@property
|
||
def path(self) -> Path:
|
||
return self._path
|
||
|
||
def _write_header(self, zong_pai_hao: str, mode: str) -> None:
|
||
self._lines.append("=" * 70)
|
||
self._lines.append(f"总排号: {zong_pai_hao}")
|
||
self._lines.append(f"分类模式: {mode}")
|
||
self._lines.append(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
self._lines.append("=" * 70)
|
||
self._flush()
|
||
|
||
def log_param_text(self, param_text: str | None) -> None:
|
||
self._lines.append("")
|
||
self._lines.append("--- 新参数原文 ---")
|
||
self._lines.append(param_text if param_text else "(空)")
|
||
self._flush()
|
||
|
||
def log_llm_attempt(
|
||
self,
|
||
attempt_no: int,
|
||
call_result: LLMCallResult,
|
||
parsed: ParsedResult | None,
|
||
format_error: str | None,
|
||
) -> None:
|
||
"""记录一次 LLM 调用的完整往返，无论成功还是失败都调用本方法。
|
||
|
||
call_result.error 非 None 表示网络/接口层面调用失败（对话都没发起成功
|
||
或没拿到回复）；call_result.error 为 None 但 format_error 非 None 表示
|
||
调用成功但模型输出格式不合规；两者都为 None 表示本次调用完全成功。
|
||
"""
|
||
self._lines.append("")
|
||
self._lines.append(f"--- 第 {attempt_no} 次尝试 ---")
|
||
|
||
self._lines.append("")
|
||
self._lines.append("[发送给模型的完整对话]")
|
||
for msg in call_result.messages:
|
||
role = msg["role"]
|
||
content = msg["content"]
|
||
self._lines.append(f" [{role}]")
|
||
for line in content.splitlines():
|
||
self._lines.append(f" {line}")
|
||
|
||
self._lines.append("")
|
||
if call_result.error is not None:
|
||
self._lines.append("[调用结果] 网络/接口调用失败")
|
||
self._lines.append(f" 错误信息: {call_result.error}")
|
||
else:
|
||
self._lines.append("[调用结果] 已收到模型回复")
|
||
self._lines.append("[模型原始回复]")
|
||
for line in (call_result.raw_response or "").splitlines():
|
||
self._lines.append(f" {line}")
|
||
|
||
self._lines.append("")
|
||
if format_error is not None:
|
||
self._lines.append("[格式校验] 失败")
|
||
self._lines.append(f" 原因: {format_error}")
|
||
else:
|
||
self._lines.append("[格式校验] 通过")
|
||
if parsed is not None:
|
||
self._lines.append(f" has_attachment = {parsed.has_attachment}")
|
||
self._lines.append(f" types = {parsed.types}")
|
||
self._flush()
|
||
|
||
def log_final_result(self, status: str, parsed: ParsedResult | None) -> None:
|
||
self._lines.append("")
|
||
self._lines.append("=" * 70)
|
||
self._lines.append(f"最终状态: {status}")
|
||
if parsed is not None:
|
||
self._lines.append(f"最终结果: has_attachment={parsed.has_attachment}, types={parsed.types}")
|
||
self._lines.append(f"结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
self._lines.append("=" * 70)
|
||
self._flush()
|
||
|
||
def _flush(self) -> None:
|
||
"""每次追加内容后立即落盘，避免进程中途异常退出导致日志丢失。"""
|
||
with self._path.open("a", encoding="utf-8") as f:
|
||
f.write("\n".join(self._lines) + "\n")
|
||
self._lines = []
|
||
|
||
def close(self) -> None:
|
||
"""预留的显式收尾方法；当前实现下 _flush 已即时写盘，close 是空操作，
|
||
但保留该接口方便未来切换为缓冲写入而不改动调用方代码。"""
|
||
pass
|