Procházet zdrojové kódy

标准模块增加采集日志

wangyang před 5 dny
rodič
revize
88867cae90
2 změnil soubory, kde provedl 76 přidání a 1 odebrání
  1. 75 0
      llmops/agents/data_stardard.py
  2. 1 1
      llmops/folder_scanner.py

+ 75 - 0
llmops/agents/data_stardard.py

@@ -5,6 +5,8 @@ import io
 import csv
 import datetime
 import httpx
+import json
+import uuid
 
 # --- LangChain Imports ---
 from langchain_openai import ChatOpenAI
@@ -34,6 +36,9 @@ class TransactionParserAgent:
         # 定义 JSON 解析器
         self.parser = JsonOutputParser()
 
+        # 初始化API调用跟踪
+        self.api_calls = []
+
     async def _invoke_miner_u(self, file_path: str) -> str:
         """调用 MinerU 并提取纯行数据 (保持 httpx 调用不变,因为这不是 LLM)"""
         miner_start_time = time.perf_counter()
@@ -174,11 +179,81 @@ JSON Array:
             try:
                 # --- LangChain 调用 ---
                 # 使用 ainvoke 异步调用链
+                # 记录API调用开始时间
+                call_start_time = datetime.datetime.now()
+
                 data_data = await chain.ainvoke({
                     "start_id": global_tx_counter,
                     "chunk_data": chunk_str
                 })
 
+                # 记录API调用结束时间
+                call_end_time = datetime.datetime.now()
+
+                # 记录API调用结果 - 简化版:只保存提示词和结果数据
+                call_id = f"api_llm_数据转换_{'{:.2f}'.format((call_end_time - call_start_time).total_seconds())}"
+
+                # 从chain中提取提示词(如果可能)
+                prompt_content = ""
+                try:
+                    # 尝试从chain获取最后的消息内容
+                    if hasattr(chain, 'get_prompts'):
+                        prompts = chain.get_prompts()
+                        if prompts:
+                            prompt_content = str(prompts[-1])
+                    else:
+                        # 如果无法获取,构造基本的提示词信息
+                        prompt_content = f"转换批次数据,start_id: {global_tx_counter}, chunk_data: {chunk_str[:200]}..."
+                except:
+                    prompt_content = f"转换批次数据,start_id: {global_tx_counter}, chunk_data: {chunk_str[:200]}..."
+
+                api_call_info = {
+                    "call_id": call_id,
+                    "start_time": call_start_time.isoformat(),
+                    "end_time": call_end_time.isoformat(),
+                    "duration": (call_end_time - call_start_time).total_seconds(),
+                    "prompt": prompt_content,
+                    "input_params": {
+                        "start_id": global_tx_counter,
+                        "chunk_data": chunk_str
+                    },
+                    "llm_result": data_data
+                }
+                self.api_calls.append(api_call_info)
+
+                # 保存API结果到文件 (Markdown格式,更易阅读)
+                # 使用运行ID创建独立的文件夹
+                run_id = os.environ.get('FLOW_RUN_ID', 'default')
+                api_results_dir = f"api_results_{run_id}"
+                os.makedirs(api_results_dir, exist_ok=True)
+                timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+                filename = f"{timestamp}_{call_id}.md"
+                filepath = os.path.join(api_results_dir, filename)
+
+                try:
+                    with open(filepath, 'w', encoding='utf-8') as f:
+                        f.write("# 数据转换结果\n\n")
+                        f.write("## 调用信息\n\n")
+                        f.write(f"- 调用ID: {call_id}\n")
+                        f.write(f"- 开始时间: {call_start_time.isoformat()}\n")
+                        f.write(f"- 结束时间: {call_end_time.isoformat()}\n")
+                        f.write(f"- 执行时长: {(call_end_time - call_start_time).total_seconds():.2f} 秒\n")
+                        f.write("\n## 提示词入参\n\n")
+                        f.write("```\n")
+                        f.write(api_call_info["prompt"])
+                        f.write("\n```\n\n")
+                        f.write("## 输入参数\n\n")
+                        f.write("```json\n")
+                        f.write(json.dumps(api_call_info["input_params"], ensure_ascii=False, indent=2))
+                        f.write("\n```\n\n")
+                        f.write("## LLM返回结果\n\n")
+                        f.write("```json\n")
+                        f.write(json.dumps(api_call_info["llm_result"], ensure_ascii=False, indent=2))
+                        f.write("\n```\n")
+                    print(f"[API_RESULT] 保存API结果文件: {filepath}")
+                except Exception as e:
+                    print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
+
                 # print(f"💡 LLM 返回数据: {data_data}")
 
                 # 兼容处理:LangChain Parser 通常会直接返回 List 或 Dict

+ 1 - 1
llmops/folder_scanner.py

@@ -335,7 +335,7 @@ def main():
 
     # 配置期望值
     config = ScanConfig(
-        expected_decision_count=3,  # 期望每个文件夹有3个决策agent文件
+        expected_decision_count=-1,  # 期望每个文件夹有3个决策agent文件
         expected_outline_count=1,  # 期望每个文件夹有1个大纲生成agent文件
         expected_metric_count=20  # 期望每个文件夹有20个指标计算agent文件
     )