Quellcode durchsuchen

全流程批量执行

jiaqiang vor 5 Tagen
Ursprung
Commit
23b2d0e9df
1 geänderte Dateien mit 44 neuen und 7 gelöschten Zeilen
  1. 44 7
      llmops/batch_runner_all.py

+ 44 - 7
llmops/batch_runner_all.py

@@ -38,6 +38,12 @@ current_dir = os.path.dirname(os.path.abspath(__file__))
 parent_dir = os.path.dirname(current_dir)
 sys.path.insert(0, parent_dir)
 
+
+os.environ["LANGCHAIN_TRACING_V2"] = "false"
+os.environ["LANGCHAIN_API_KEY"] = ""
+# 禁用 LangGraph 的追踪
+os.environ["LANGSMITH_TRACING"] = "false"
+
 # 根据执行方式选择导入方式
 if __name__ == "__main__":
     # 直接执行文件时,使用绝对导入
@@ -53,6 +59,7 @@ import config
 # ========== 配置参数 ==========
 RUNS = 10  # 运行次数
 INDUSTRY = "农业"  # 行业
+ORIGINAL_DATA_FILE="data_files/11111.png"   # 原始测试数据文件PDF
 DATA_FILE = "data_files/交易流水样例数据.csv"  # 数据文件路径
 QUESTION = "请生成一份详细的农业经营贷流水分析报告,需要包含:1.总收入和总支出统计 2.收入笔数和支出笔数 3.各类型收入支出占比分析 4.交易对手收入支出TOP3排名 5.按月份的收入支出趋势分析 6.账户数量和交易时间范围统计 7.资金流入流出月度统计等全面指标"  # 分析查询
 # ==============================
@@ -109,6 +116,35 @@ async def run_single_flow(run_id: str, question: str, industry: str, data: List[
         }
 
 
+async def data_standardize():
+    """
+    提取数据,进行标准化处理
+    """
+    from llmops.agents.data_stardard import TransactionParserAgent
+    from llmops.config import DEEPSEEK_API_KEY, multimodal_api_url
+
+    # 数据标准化agent
+    standard_agent = TransactionParserAgent(
+        api_key=DEEPSEEK_API_KEY,
+        multimodal_api_url=multimodal_api_url
+    )
+    # 执行解析
+    full_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ORIGINAL_DATA_FILE)
+    try:
+        result = await standard_agent.run_workflow_task(full_path)
+        if result["status"] == "success":
+            print(f"🎯 数据标准化任务完成!")
+            # 标准化后的文件
+            standard_file_path = result['file_path']
+            return standard_file_path
+        else:
+            print(f"❌ 数据标准化任务失败: {result['message']}")
+            raise ValueError("数据标准化异常")
+    except Exception as e:
+        print(f"数据标准化处理异常:{e}")
+        raise e
+
+
 async def run_batch(runs: int, question: str, industry: str, data_file: str):
     """
     批量运行工作流
@@ -117,7 +153,7 @@ async def run_batch(runs: int, question: str, industry: str, data_file: str):
         runs: 运行次数
         question: 用户查询
         industry: 行业
-        data_file: 数据文件路径
+        data_file: 原始数据文件路径,格式支持pdf/img/csv
     """
     print("🚀 批量运行器启动")
     print(f"📊 计划运行次数: {runs}")
@@ -131,11 +167,12 @@ async def run_batch(runs: int, question: str, industry: str, data_file: str):
         print("❌ 未找到API密钥,请检查config.py")
         return
 
-    # 加载数据
     try:
-        # 如果是相对路径,从项目根目录查找
-        if not os.path.isabs(data_file):
-            data_file = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), data_file)
+        # 执行数据标准化
+        standard_file_path = await data_standardize()
+        print(f"标准化后文件:{standard_file_path}")
+        data_file = standard_file_path
+        # 标准化数据加载
         data = DataManager.load_data_from_csv_file(data_file)
         print(f"📊 数据加载成功: {len(data)} 条记录")
     except Exception as e:
@@ -196,7 +233,7 @@ def main():
     print("🚀 使用配置参数运行批量任务")
     print(f"📊 运行次数: {RUNS}")
     print(f"🏭 行业: {INDUSTRY}")
-    print(f"📁 数据文件: {DATA_FILE}")
+    print(f"📁 数据文件: {ORIGINAL_DATA_FILE}")
     print(f"❓ 查询: {QUESTION[:50]}...")
     print("-" * 80)
 
@@ -205,7 +242,7 @@ def main():
         runs=RUNS,
         question=QUESTION,
         industry=INDUSTRY,
-        data_file=DATA_FILE
+        data_file=ORIGINAL_DATA_FILE
     ))