116 lines
2.8 KiB
Python
116 lines
2.8 KiB
Python
"""
|
|
因果分析日志记录器
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class CausalAnalysisLogger:
|
|
"""因果分析日志记录器"""
|
|
|
|
def __init__(self, log_path: str = "causal_analysis.log.md"):
|
|
self.log_path = log_path
|
|
self.log_entries: List[Dict[str, Any]] = []
|
|
self._init_log_file()
|
|
self._load_existing_entries()
|
|
|
|
def _init_log_file(self):
|
|
"""初始化日志文件"""
|
|
header = """# 因果分析日志
|
|
|
|
## 日志说明
|
|
本文档记录因果推断分析的所有输入参数和输出结果。
|
|
|
|
## 分析记录
|
|
"""
|
|
if not os.path.exists(self.log_path):
|
|
with open(self.log_path, "w", encoding="utf-8") as f:
|
|
f.write(header)
|
|
|
|
def _load_existing_entries(self):
|
|
"""加载已存在的分析记录数量"""
|
|
if not os.path.exists(self.log_path):
|
|
self.analysis_count = 0
|
|
return
|
|
|
|
analysis_count = 0
|
|
with open(self.log_path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
matches = re.findall(r"### 分析 #(\d+)", content)
|
|
if matches:
|
|
analysis_count = max(int(m) for m in matches)
|
|
self.analysis_count = analysis_count
|
|
|
|
def _append_to_log(self, content: str):
|
|
"""追加内容到日志文件"""
|
|
with open(self.log_path, "a", encoding="utf-8") as f:
|
|
f.write(content)
|
|
|
|
def log_analysis(
|
|
self,
|
|
system_prompt: str,
|
|
user_prompt: str,
|
|
model_output: str,
|
|
parameters: Dict[str, Any],
|
|
report: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""记录一次完整的分析过程"""
|
|
self.analysis_count += 1
|
|
analysis_id = f"{self.analysis_count:03d}"
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
report_json = ""
|
|
if report is not None:
|
|
report_json = (
|
|
f"\n#### 分析报告\n"
|
|
f"```json\n{json.dumps(report, indent=2, ensure_ascii=False)}\n```\n"
|
|
)
|
|
|
|
entry = f"""
|
|
---
|
|
|
|
### 分析 #{analysis_id}
|
|
**时间**: {timestamp}
|
|
|
|
#### 系统提示词
|
|
```
|
|
{system_prompt}
|
|
```
|
|
|
|
#### 用户提示词
|
|
```
|
|
{user_prompt}
|
|
```
|
|
|
|
#### LLM 输出
|
|
```
|
|
{model_output}
|
|
```
|
|
{report_json}
|
|
#### 调用参数
|
|
```json
|
|
{json.dumps(parameters, indent=2, ensure_ascii=False)}
|
|
```
|
|
|
|
---
|
|
|
|
"""
|
|
self._append_to_log(entry)
|
|
self.log_entries.append(
|
|
{
|
|
"id": analysis_id,
|
|
"timestamp": timestamp,
|
|
"system_prompt": system_prompt,
|
|
"user_prompt": user_prompt,
|
|
"model_output": model_output,
|
|
"parameters": parameters,
|
|
"report": report,
|
|
}
|
|
)
|
|
|
|
print(f"分析 #{analysis_id} 已记录到 {self.log_path}")
|