|
|
@@ -0,0 +1,670 @@
|
|
|
+"""
|
|
|
+指标计算Agent (Metric Calculation Agent)
|
|
|
+====================================
|
|
|
+
|
|
|
+此Agent负责根据意图识别结果执行具体的指标计算任务。
|
|
|
+
|
|
|
+核心功能：
|
|
|
+1. 配置文件加载：读取和解析JSON格式的指标计算配置文件
|
|
|
+2. API调用管理：根据配置文件调用相应的计算API
|
|
|
+3. 结果处理：处理API返回的数据，提取关键指标
|
|
|
+4. 错误处理：处理API调用失败、网络异常等错误情况
|
|
|
+5. 结果验证：验证计算结果的合理性和完整性
|
|
|
+
|
|
|
+工作流程：
|
|
|
+1. 接收意图识别结果和用户参数
|
|
|
+2. 加载对应的指标计算配置文件
|
|
|
+3. 构造API请求参数
|
|
|
+4. 调用远程计算服务
|
|
|
+5. 解析和验证返回结果
|
|
|
+6. 返回结构化的计算结果
|
|
|
+
|
|
|
+技术实现：
|
|
|
+- 支持动态加载JSON配置文件
|
|
|
+- 使用requests库进行HTTP API调用
|
|
|
+- 集成LangChain用于复杂计算逻辑（可选）
|
|
|
+- 完善的错误处理和超时机制
|
|
|
+- 支持多种计算方法（标准、高级、自定义）
|
|
|
+
|
|
|
+配置文件结构：
|
|
|
+- api_config: API端点和认证信息
|
|
|
+- param_mapping: 参数映射规则
|
|
|
+- input_schema: 输入数据验证规则
|
|
|
+- output_schema: 输出数据结构定义
|
|
|
+- calculation_logic: 计算逻辑描述
|
|
|
+
|
|
|
+作者: Big Agent Team
|
|
|
+版本: 1.0.0
|
|
|
+创建时间: 2024-12-18
|
|
|
+"""
|
|
|
+
|
|
|
+import os
|
|
|
+import json
|
|
|
+import requests
|
|
|
+from datetime import datetime
|
|
|
+from typing import Dict, List, Any, Optional
|
|
|
+from langchain_openai import ChatOpenAI
|
|
|
+from langchain_core.prompts import ChatPromptTemplate
|
|
|
+import re
|
|
|
+
|
|
|
+
|
|
|
+class MetricCalculationAgent:
|
|
|
+ """远程指标计算Agent"""
|
|
|
+
|
|
|
+ def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
|
|
|
+ """
|
|
|
+ 初始化指标计算Agent
|
|
|
+
|
|
|
+ Args:
|
|
|
+ api_key: DeepSeek API密钥
|
|
|
+ base_url: DeepSeek API基础URL
|
|
|
+ """
|
|
|
+ self.llm = ChatOpenAI(
|
|
|
+ model="deepseek-chat",
|
|
|
+ api_key=api_key,
|
|
|
+ base_url=base_url,
|
|
|
+ temperature=0.1
|
|
|
+ )
|
|
|
+
|
|
|
+ # 加载配置文件
|
|
|
+ self.configs = self._load_configs()
|
|
|
+
|
|
|
+ # 加载数据文件映射
|
|
|
+ self.data_files = self._load_data_files()
|
|
|
+
|
|
|
+ # 初始化API调用跟踪
|
|
|
+ self.api_calls = []
|
|
|
+
|
|
|
+ def _load_data_files(self) -> Dict[str, str]:
|
|
|
+ """加载数据文件映射"""
|
|
|
+ data_files = {}
|
|
|
+ data_dir = "data_files"
|
|
|
+
|
|
|
+ if os.path.exists(data_dir):
|
|
|
+ for file in os.listdir(data_dir):
|
|
|
+ if file.endswith('.json'):
|
|
|
+ try:
|
|
|
+ # 提取文件名中的关键词，用于匹配配置文件
|
|
|
+ key = file.replace('原始数据-流水分析-', '').replace('.json', '')
|
|
|
+ data_files[key] = os.path.join(data_dir, file)
|
|
|
+ except Exception as e:
|
|
|
+ print(f"处理数据文件 {file} 失败: {e}")
|
|
|
+
|
|
|
+ return data_files
|
|
|
+
|
|
|
+ def _select_data_file(self, config_name: str) -> Optional[str]:
|
|
|
+ """
|
|
|
+ 根据配置文件名选择对应的数据文件
|
|
|
+
|
|
|
+ Args:
|
|
|
+ config_name: 配置文件名
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 数据文件路径，如果找不到则返回None
|
|
|
+ """
|
|
|
+ # 配置文件名模式：指标计算-{category}-{metric}.json
|
|
|
+ # 数据文件名模式：原始数据-流水分析-{category}.json
|
|
|
+
|
|
|
+ # 从配置文件名中提取类别信息
|
|
|
+ match = re.search(r'指标计算-(.+?)-', config_name)
|
|
|
+ if match:
|
|
|
+ category = match.group(1)
|
|
|
+
|
|
|
+ # 优先选择原始数据文件
|
|
|
+ # 1. 首先查找完全匹配的原始数据文件
|
|
|
+ if category in self.data_files:
|
|
|
+ file_path = self.data_files[category]
|
|
|
+ if '原始数据' in file_path:
|
|
|
+ return file_path
|
|
|
+
|
|
|
+ # 2. 如果没有完全匹配，查找包含类别的原始数据文件
|
|
|
+ for key, file_path in self.data_files.items():
|
|
|
+ if category in key and '原始数据' in file_path:
|
|
|
+ return file_path
|
|
|
+
|
|
|
+ # 如果找不到匹配的文件，返回默认的农业原始数据文件（如果存在）
|
|
|
+ if '农业' in self.data_files:
|
|
|
+ return self.data_files['农业']
|
|
|
+
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _load_table_data(self, data_file_path: str) -> List[Dict]:
|
|
|
+ """加载数据文件中的表格数据"""
|
|
|
+ try:
|
|
|
+ with open(data_file_path, 'r', encoding='utf-8') as f:
|
|
|
+ data = json.load(f)
|
|
|
+ return data if isinstance(data, list) else []
|
|
|
+ except Exception as e:
|
|
|
+ print(f"加载数据文件 {data_file_path} 失败: {e}")
|
|
|
+ return []
|
|
|
+
|
|
|
+ def _load_configs(self) -> Dict[str, Dict]:
|
|
|
+ """加载所有配置文件"""
|
|
|
+ configs = {}
|
|
|
+ json_dir = "json_files"
|
|
|
+
|
|
|
+ if os.path.exists(json_dir):
|
|
|
+ for file in os.listdir(json_dir):
|
|
|
+ if file.endswith('.json'):
|
|
|
+ try:
|
|
|
+ with open(os.path.join(json_dir, file), 'r', encoding='utf-8') as f:
|
|
|
+ config = json.load(f)
|
|
|
+ key = file.replace('.json', '')
|
|
|
+ configs[key] = config
|
|
|
+ except Exception as e:
|
|
|
+ print(f"加载配置文件 {file} 失败: {e}")
|
|
|
+
|
|
|
+ return configs
|
|
|
+
|
|
|
+ async def calculate_metrics(self, intent_result: Dict[str, Any]) -> Dict[str, Any]:
|
|
|
+ """
|
|
|
+ 根据意图识别结果进行指标计算
|
|
|
+
|
|
|
+ Args:
|
|
|
+ intent_result: 意图识别结果
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 指标计算结果
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ results = []
|
|
|
+ target_configs = intent_result.get("target_configs", [])
|
|
|
+
|
|
|
+ if not target_configs:
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": "没有找到需要调用的配置文件",
|
|
|
+ "results": []
|
|
|
+ }
|
|
|
+
|
|
|
+ for config_name in target_configs:
|
|
|
+ if config_name in self.configs:
|
|
|
+ config = self.configs[config_name]
|
|
|
+ result = await self._call_metric_api(config, intent_result, config_name)
|
|
|
+ results.append({
|
|
|
+ "config_name": config_name,
|
|
|
+ "result": result
|
|
|
+ })
|
|
|
+ else:
|
|
|
+ results.append({
|
|
|
+ "config_name": config_name,
|
|
|
+ "error": f"配置文件 {config_name} 不存在"
|
|
|
+ })
|
|
|
+
|
|
|
+ return {
|
|
|
+ "success": True,
|
|
|
+ "results": results,
|
|
|
+ "total_configs": len(target_configs),
|
|
|
+ "successful_calculations": len([r for r in results if "result" in r])
|
|
|
+ }
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ print(f"指标计算失败: {e}")
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": f"指标计算过程中发生错误: {str(e)}",
|
|
|
+ "results": []
|
|
|
+ }
|
|
|
+
|
|
|
+ async def _call_metric_api(self, config: Dict[str, Any], intent_result: Dict[str, Any], config_name: str) -> Dict[str, Any]:
|
|
|
+ """
|
|
|
+ 调用具体的指标计算API
|
|
|
+
|
|
|
+ Args:
|
|
|
+ config: 配置文件
|
|
|
+ intent_result: 意图识别结果
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ API调用结果
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ # 记录API调用开始
|
|
|
+ start_time = datetime.now()
|
|
|
+
|
|
|
+ # 使用真实API服务的配置
|
|
|
+ method = "POST"
|
|
|
+ url = "http://10.192.72.11:6300/api/data_analyst/full" # 真实API服务地址
|
|
|
+ headers = {
|
|
|
+ "Accept": "*/*",
|
|
|
+ "Accept-Encoding": "gzip, deflate, br",
|
|
|
+ "Connection": "keep-alive",
|
|
|
+ "Content-Type": "application/json",
|
|
|
+ "User-Agent": "PostmanRuntime-ApipostRuntime/1.1.0"
|
|
|
+ }
|
|
|
+ timeout = 600 # 10分钟超时，给慢API更多时间
|
|
|
+
|
|
|
+ # 添加重试机制，最多重试3次
|
|
|
+ max_retries = 3
|
|
|
+ retry_delay = 5 # 每次重试间隔5秒
|
|
|
+
|
|
|
+ for attempt in range(max_retries):
|
|
|
+ try:
|
|
|
+ print(f"🔄 API调用尝试 {attempt + 1}/{max_retries} (配置: {config_name})")
|
|
|
+
|
|
|
+ # 准备请求数据
|
|
|
+ request_data = self._prepare_request_data(config, intent_result, config_name)
|
|
|
+
|
|
|
+ # 根据HTTP方法调用API
|
|
|
+ if method.upper() == "GET":
|
|
|
+ params = request_data.get("params", {})
|
|
|
+ response = requests.get(url, headers=headers, params=params, timeout=timeout)
|
|
|
+ elif method.upper() == "POST":
|
|
|
+ json_data = request_data.get("json", {})
|
|
|
+ response = requests.post(url, headers=headers, json=json_data, timeout=timeout)
|
|
|
+ else:
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": f"不支持的HTTP方法: {method}"
|
|
|
+ }
|
|
|
+
|
|
|
+ # 处理响应
|
|
|
+ if response.status_code == 200:
|
|
|
+ try:
|
|
|
+ response_data = response.json()
|
|
|
+
|
|
|
+ # 检查API响应结构并提取结果
|
|
|
+ extracted_result = None
|
|
|
+ if isinstance(response_data, dict):
|
|
|
+ # 检查是否有code字段和data.result结构
|
|
|
+ if response_data.get("code") == 0 and "data" in response_data:
|
|
|
+ data = response_data["data"]
|
|
|
+ if "result" in data:
|
|
|
+ # 从result字段中提取JSON
|
|
|
+ extracted_result = self._extract_json_from_result(data["result"])
|
|
|
+
|
|
|
+ # 记录API调用结果
|
|
|
+ end_time = datetime.now()
|
|
|
+ call_id = f"api_{config_name}_{'{:.2f}'.format((end_time - start_time).total_seconds())}"
|
|
|
+ api_call_info = {
|
|
|
+ "call_id": call_id,
|
|
|
+ "timestamp": end_time.isoformat(),
|
|
|
+ "agent": "MetricCalculationAgent",
|
|
|
+ "api_endpoint": url,
|
|
|
+ "config_name": config_name,
|
|
|
+ "request": {
|
|
|
+ "method": method,
|
|
|
+ "url": url,
|
|
|
+ "headers": headers,
|
|
|
+ "json_data": json_data if method.upper() == "POST" else None,
|
|
|
+ "params": params if method.upper() == "GET" else None,
|
|
|
+ "start_time": start_time.isoformat()
|
|
|
+ },
|
|
|
+ "response": {
|
|
|
+ "status_code": response.status_code,
|
|
|
+ "data": response_data,
|
|
|
+ "extracted_result": extracted_result,
|
|
|
+ "end_time": end_time.isoformat(),
|
|
|
+ "duration": (end_time - start_time).total_seconds()
|
|
|
+ },
|
|
|
+ "success": True
|
|
|
+ }
|
|
|
+ self.api_calls.append(api_call_info)
|
|
|
+
|
|
|
+ # 保存API结果到文件
|
|
|
+ api_results_dir = "api_results"
|
|
|
+ os.makedirs(api_results_dir, exist_ok=True)
|
|
|
+ filename = f"{call_id}.json"
|
|
|
+ filepath = os.path.join(api_results_dir, filename)
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(filepath, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(api_call_info, f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"[API_RESULT] 保存API结果文件: {filepath}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
|
|
|
+
|
|
|
+ return {
|
|
|
+ "success": True,
|
|
|
+ "data": response_data,
|
|
|
+ "extracted_result": extracted_result,
|
|
|
+ "status_code": response.status_code
|
|
|
+ }
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ # 记录API调用结果（JSON解析失败）
|
|
|
+ end_time = datetime.now()
|
|
|
+ call_id = f"api_{config_name}_{'{:.2f}'.format((end_time - start_time).total_seconds())}"
|
|
|
+ api_call_info = {
|
|
|
+ "call_id": call_id,
|
|
|
+ "timestamp": end_time.isoformat(),
|
|
|
+ "agent": "MetricCalculationAgent",
|
|
|
+ "api_endpoint": url,
|
|
|
+ "config_name": config_name,
|
|
|
+ "request": {
|
|
|
+ "method": method,
|
|
|
+ "url": url,
|
|
|
+ "headers": headers,
|
|
|
+ "json_data": json_data if method.upper() == "POST" else None,
|
|
|
+ "params": params if method.upper() == "GET" else None,
|
|
|
+ "start_time": start_time.isoformat()
|
|
|
+ },
|
|
|
+ "response": {
|
|
|
+ "status_code": response.status_code,
|
|
|
+ "data": response.text,
|
|
|
+ "error": "JSON解析失败",
|
|
|
+ "end_time": end_time.isoformat(),
|
|
|
+ "duration": (end_time - start_time).total_seconds()
|
|
|
+ },
|
|
|
+ "success": False
|
|
|
+ }
|
|
|
+ self.api_calls.append(api_call_info)
|
|
|
+
|
|
|
+ # 保存API结果到文件
|
|
|
+ api_results_dir = "api_results"
|
|
|
+ os.makedirs(api_results_dir, exist_ok=True)
|
|
|
+ filename = f"{call_id}.json"
|
|
|
+ filepath = os.path.join(api_results_dir, filename)
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(filepath, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(api_call_info, f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"[API_RESULT] 保存API结果文件: {filepath}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
|
|
|
+
|
|
|
+ return {
|
|
|
+ "success": True,
|
|
|
+ "data": response.text,
|
|
|
+ "extracted_result": None,
|
|
|
+ "status_code": response.status_code
|
|
|
+ }
|
|
|
+ else:
|
|
|
+ # 记录API调用结果（HTTP错误）
|
|
|
+ end_time = datetime.now()
|
|
|
+ call_id = f"api_{config_name}_{'{:.2f}'.format((end_time - start_time).total_seconds())}"
|
|
|
+ api_call_info = {
|
|
|
+ "call_id": call_id,
|
|
|
+ "timestamp": end_time.isoformat(),
|
|
|
+ "agent": "MetricCalculationAgent",
|
|
|
+ "api_endpoint": url,
|
|
|
+ "config_name": config_name,
|
|
|
+ "request": {
|
|
|
+ "method": method,
|
|
|
+ "url": url,
|
|
|
+ "headers": headers,
|
|
|
+ "json_data": json_data if method.upper() == "POST" else None,
|
|
|
+ "params": params if method.upper() == "GET" else None,
|
|
|
+ "start_time": start_time.isoformat()
|
|
|
+ },
|
|
|
+ "response": {
|
|
|
+ "status_code": response.status_code,
|
|
|
+ "error": response.text,
|
|
|
+ "end_time": end_time.isoformat(),
|
|
|
+ "duration": (end_time - start_time).total_seconds()
|
|
|
+ },
|
|
|
+ "success": False
|
|
|
+ }
|
|
|
+ self.api_calls.append(api_call_info)
|
|
|
+
|
|
|
+ # 保存API结果到文件
|
|
|
+ api_results_dir = "api_results"
|
|
|
+ os.makedirs(api_results_dir, exist_ok=True)
|
|
|
+ filename = f"{call_id}.json"
|
|
|
+ filepath = os.path.join(api_results_dir, filename)
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(filepath, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(api_call_info, f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"[API_RESULT] 保存API结果文件: {filepath}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
|
|
|
+
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": f"API调用失败，状态码: {response.status_code}",
|
|
|
+ "response": response.text
|
|
|
+ }
|
|
|
+
|
|
|
+ # 如果执行到这里，说明本次尝试成功，跳出重试循环
|
|
|
+ break
|
|
|
+
|
|
|
+ except requests.exceptions.Timeout:
|
|
|
+ # 记录API调用结果（超时）
|
|
|
+ end_time = datetime.now()
|
|
|
+ call_id = f"api_{config_name}_{'{:.2f}'.format((end_time - start_time).total_seconds())}"
|
|
|
+ api_call_info = {
|
|
|
+ "call_id": call_id,
|
|
|
+ "timestamp": end_time.isoformat(),
|
|
|
+ "agent": "MetricCalculationAgent",
|
|
|
+ "api_endpoint": url,
|
|
|
+ "config_name": config_name,
|
|
|
+ "request": {
|
|
|
+ "method": method,
|
|
|
+ "url": url,
|
|
|
+ "headers": headers,
|
|
|
+ "json_data": json_data if method.upper() == "POST" else None,
|
|
|
+ "params": params if method.upper() == "GET" else None,
|
|
|
+ "start_time": start_time.isoformat()
|
|
|
+ },
|
|
|
+ "response": {
|
|
|
+ "error": "API调用超时",
|
|
|
+ "end_time": end_time.isoformat(),
|
|
|
+ "duration": (end_time - start_time).total_seconds()
|
|
|
+ },
|
|
|
+ "success": False
|
|
|
+ }
|
|
|
+ self.api_calls.append(api_call_info)
|
|
|
+
|
|
|
+ # 保存API结果到文件
|
|
|
+ api_results_dir = "api_results"
|
|
|
+ os.makedirs(api_results_dir, exist_ok=True)
|
|
|
+ filename = f"{call_id}.json"
|
|
|
+ filepath = os.path.join(api_results_dir, filename)
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(filepath, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(api_call_info, f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"[API_RESULT] 保存API结果文件: {filepath}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
|
|
|
+
|
|
|
+ # 如果不是最后一次尝试，等待后重试
|
|
|
+ if attempt < max_retries - 1:
|
|
|
+ print(f"⏳ API调用超时，{retry_delay}秒后重试...")
|
|
|
+ import time
|
|
|
+ time.sleep(retry_delay)
|
|
|
+ continue
|
|
|
+ else:
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": "API调用超时"
|
|
|
+ }
|
|
|
+ except requests.exceptions.RequestException as e:
|
|
|
+ # 记录API调用结果（请求异常）
|
|
|
+ end_time = datetime.now()
|
|
|
+ call_id = f"api_{config_name}_{'{:.2f}'.format((end_time - start_time).total_seconds())}"
|
|
|
+ api_call_info = {
|
|
|
+ "call_id": call_id,
|
|
|
+ "timestamp": end_time.isoformat(),
|
|
|
+ "agent": "MetricCalculationAgent",
|
|
|
+ "api_endpoint": url,
|
|
|
+ "config_name": config_name,
|
|
|
+ "request": {
|
|
|
+ "method": method,
|
|
|
+ "url": url,
|
|
|
+ "headers": headers,
|
|
|
+ "json_data": json_data if method.upper() == "POST" else None,
|
|
|
+ "params": params if method.upper() == "GET" else None,
|
|
|
+ "start_time": start_time.isoformat()
|
|
|
+ },
|
|
|
+ "response": {
|
|
|
+ "error": str(e),
|
|
|
+ "end_time": end_time.isoformat(),
|
|
|
+ "duration": (end_time - start_time).total_seconds()
|
|
|
+ },
|
|
|
+ "success": False
|
|
|
+ }
|
|
|
+ self.api_calls.append(api_call_info)
|
|
|
+
|
|
|
+ # 保存API结果到文件
|
|
|
+ api_results_dir = "api_results"
|
|
|
+ os.makedirs(api_results_dir, exist_ok=True)
|
|
|
+ filename = f"{call_id}.json"
|
|
|
+ filepath = os.path.join(api_results_dir, filename)
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(filepath, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(api_call_info, f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"[API_RESULT] 保存API结果文件: {filepath}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
|
|
|
+
|
|
|
+ # 如果不是最后一次尝试，等待后重试
|
|
|
+ if attempt < max_retries - 1:
|
|
|
+ print(f"❌ API调用异常: {str(e)}，{retry_delay}秒后重试...")
|
|
|
+ import time
|
|
|
+ time.sleep(retry_delay)
|
|
|
+ continue
|
|
|
+ else:
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": f"API调用异常: {str(e)}"
|
|
|
+ }
|
|
|
+ except Exception as e:
|
|
|
+ # 记录API调用结果（其他异常）
|
|
|
+ end_time = datetime.now()
|
|
|
+ call_id = f"api_{config_name}_{'{:.2f}'.format((end_time - start_time).total_seconds())}"
|
|
|
+ api_call_info = {
|
|
|
+ "call_id": call_id,
|
|
|
+ "timestamp": end_time.isoformat(),
|
|
|
+ "agent": "MetricCalculationAgent",
|
|
|
+ "api_endpoint": url,
|
|
|
+ "config_name": config_name,
|
|
|
+ "request": {
|
|
|
+ "method": method,
|
|
|
+ "url": url,
|
|
|
+ "headers": headers,
|
|
|
+ "json_data": json_data if method.upper() == "POST" else None,
|
|
|
+ "params": params if method.upper() == "GET" else None,
|
|
|
+ "start_time": start_time.isoformat()
|
|
|
+ },
|
|
|
+ "response": {
|
|
|
+ "error": str(e),
|
|
|
+ "end_time": end_time.isoformat(),
|
|
|
+ "duration": (end_time - start_time).total_seconds()
|
|
|
+ },
|
|
|
+ "success": False
|
|
|
+ }
|
|
|
+ self.api_calls.append(api_call_info)
|
|
|
+
|
|
|
+ # 保存API结果到文件
|
|
|
+ api_results_dir = "api_results"
|
|
|
+ os.makedirs(api_results_dir, exist_ok=True)
|
|
|
+ filename = f"{call_id}.json"
|
|
|
+ filepath = os.path.join(api_results_dir, filename)
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(filepath, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(api_call_info, f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"[API_RESULT] 保存API结果文件: {filepath}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
|
|
|
+
|
|
|
+ # 如果不是最后一次尝试，等待后重试
|
|
|
+ if attempt < max_retries - 1:
|
|
|
+ print(f"❌ 其他异常: {str(e)}，{retry_delay}秒后重试...")
|
|
|
+ import time
|
|
|
+ time.sleep(retry_delay)
|
|
|
+ continue
|
|
|
+ else:
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": f"API调用异常: {str(e)}"
|
|
|
+ }
|
|
|
+ except Exception as e:
|
|
|
+ # 处理所有未捕获的异常
|
|
|
+ print(f"❌ API调用过程中发生未预期的错误: {str(e)}")
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "message": f"API调用过程中发生未预期的错误: {str(e)}"
|
|
|
+ }
|
|
|
+
|
|
|
+ def _prepare_request_data(self, config: Dict[str, Any], intent_result: Dict[str, Any], config_name: str) -> Dict[str, Any]:
|
|
|
+ """
|
|
|
+ 准备API请求数据
|
|
|
+
|
|
|
+ Args:
|
|
|
+ config: 配置文件
|
|
|
+ intent_result: 意图识别结果
|
|
|
+ config_name: 配置文件名
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 请求数据
|
|
|
+ """
|
|
|
+ # 从配置文件中获取question和prompt
|
|
|
+ question = config.get("question", "")
|
|
|
+ prompt = config.get("prompt", "")
|
|
|
+
|
|
|
+ # 选择对应的数据文件
|
|
|
+ data_file_path = self._select_data_file(config_name)
|
|
|
+ table_data = []
|
|
|
+
|
|
|
+ if data_file_path:
|
|
|
+ table_data = self._load_table_data(data_file_path)
|
|
|
+ else:
|
|
|
+ print(f"警告：找不到配置文件 {config_name} 对应的数据文件")
|
|
|
+
|
|
|
+ # 构造documents数组
|
|
|
+ documents = []
|
|
|
+ if table_data:
|
|
|
+ # 使用数据文件名作为标题
|
|
|
+ title = f"数据表-{config_name}"
|
|
|
+ if data_file_path:
|
|
|
+ title = os.path.basename(data_file_path).replace('.json', '')
|
|
|
+
|
|
|
+ documents.append({
|
|
|
+ "id": 1,
|
|
|
+ "title": title,
|
|
|
+ "text": "",
|
|
|
+ "table": table_data
|
|
|
+ })
|
|
|
+
|
|
|
+ # 构造API请求体
|
|
|
+ request_data = {
|
|
|
+ "disable_planning": False,
|
|
|
+ "question": question,
|
|
|
+ "prompt": prompt,
|
|
|
+ "documents": documents
|
|
|
+ }
|
|
|
+
|
|
|
+ return {"json": request_data}
|
|
|
+
|
|
|
+ def _extract_json_from_result(self, result_text: str) -> Dict[str, Any]:
|
|
|
+ """
|
|
|
+ 从API结果文本中提取JSON内容
|
|
|
+
|
|
|
+ Args:
|
|
|
+ result_text: API返回的result字段内容
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 提取的JSON对象
|
|
|
+ """
|
|
|
+ import re
|
|
|
+ import json
|
|
|
+
|
|
|
+ try:
|
|
|
+ # 查找```json和```之间的内容
|
|
|
+ json_match = re.search(r'```json\s*(.*?)\s*```', result_text, re.DOTALL)
|
|
|
+ if json_match:
|
|
|
+ json_str = json_match.group(1).strip()
|
|
|
+ return json.loads(json_str)
|
|
|
+
|
|
|
+ # 如果没有```json标记，查找大括号包围的内容
|
|
|
+ brace_match = re.search(r'\{.*\}', result_text, re.DOTALL)
|
|
|
+ if brace_match:
|
|
|
+ json_str = brace_match.group(0).strip()
|
|
|
+ return json.loads(json_str)
|
|
|
+
|
|
|
+ # 如果都找不到，尝试直接解析整个文本
|
|
|
+ return json.loads(result_text.strip())
|
|
|
+
|
|
|
+ except json.JSONDecodeError as e:
|
|
|
+ print(f"JSON解析失败: {e}")
|
|
|
+ return {"error": f"无法解析JSON结果: {str(e)}", "raw_result": result_text}
|
|
|
+
|
|
|
+ def get_available_configs(self) -> List[str]:
|
|
|
+ """获取所有可用的配置文件名"""
|
|
|
+ return list(self.configs.keys())
|
|
|
+
|
|
|
+ def get_config_details(self, config_name: str) -> Optional[Dict]:
|
|
|
+ """获取指定配置文件的详细信息"""
|
|
|
+ return self.configs.get(config_name)
|
