LLMAsAnalyst.py

from src.utils import ConfigLoader, Neo4jHandler, DebugLogger, DataReader
import os
import json
from datetime import datetime
from src.Models import EmbeddingGenerator, LLMClient
from src import graphbuild, signalselect, signalparser
from src.sqlgenerate import SQLGenerator
from src.utils.debug_log import logger

class LLMAsAnalyst:
    def __init__(self):
        # self.query = None
        config = ConfigLoader("src/config.yaml").load_config()
        neo4j_cfg = config.get("neo4j", {})
        self.neo4j = Neo4jHandler(
            neo4j_cfg.get("uri", "bolt://localhost:7687"),
            neo4j_cfg.get("user", "neo4j"),
            neo4j_cfg.get("password", "")
        )
        self.embedding_generator = EmbeddingGenerator(self.neo4j, config)
        self.llm_client = LLMClient(config)  # 假设LLMClient按config初始化
        # self.signalselector = signalselect.SignalSelector(self.neo4j, self.embedding_generator, self.llm_client, config)
        self.graphbuilder = graphbuild.GraphBuilder(self.neo4j, self.embedding_generator, self.llm_client, config)
        self.signalparser = signalparser.SignalParser(config, self.llm_client, self.embedding_generator, self.neo4j)
        self.sqlgenerator = SQLGenerator(config, self.llm_client, self.embedding_generator)
    
    def buildgraph(self):
        """构建知识图谱"""
        self.graphbuilder.main()
    
    def signalparse(self, query):
        """解析单个查询的信号"""
        self.signalparser.main(query)
    
    def debug(self, query):
        """调试单个查询的信号解析"""
        self.signalparser.debug_main_from_step2(query)
    
    def batch_process_queries(self, 
                             input_file: str = "data/input/queries.xlsx",
                             checkpoint_dir: str = "data/output/batch_results/checkpoints") -> None:
        """
        批量处理查询文件中的所有查询语句
        
        该方法从Excel文件中读取查询列表，逐个调用signalparse进行信号解析处理。
        支持断点续跑功能，即使中途中断也可以从上次停止的地方继续执行。
        
        Args:
            input_file (str): 输入的Excel文件路径，默认为"data/input/queries.xlsx"
            checkpoint_dir (str): 断点保存目录，默认为"data/output/batch_results/checkpoints"
        
        Returns:
            None
        
        功能说明:
            1. 从Excel文件中读取query列
            2. 支持断点保存，记录已处理和失败的查询索引
            3. 对每个查询调用self.signalparse()进行处理
            4. 使用tqdm显示进度条（如果可用）
            5. 捕获并记录处理过程中的异常
            6. 支持Ctrl+C中断，并保存当前进度
        
        异常处理:
            - 读取文件失败：记录错误日志并抛出异常
            - 单个查询处理失败：记录到failed_indices，继续处理下一个
            - 键盘中断：保存进度后优雅退出
        """
        # 读取输入文件
        try:
            df = DataReader.read_excel(input_file, columns=["query"])
        except Exception as exc:
            logger.error(f"读取 {input_file} 失败: {exc}")
            raise

        # 初始化断点保存设置
        os.makedirs(checkpoint_dir, exist_ok=True)
        checkpoint_file = os.path.join(checkpoint_dir, "queries.progress.json")

        # 初始化处理状态集合
        processed_indices: set[int] = set()  # 已成功处理的查询索引
        failed_indices: set[int] = set()      # 处理失败的查询索引

        # 读取已有进度（断点恢复）
        if os.path.exists(checkpoint_file):
            try:
                with open(checkpoint_file, "r", encoding="utf-8") as f:
                    ckpt = json.load(f) or {}
                processed_indices = set(ckpt.get("processed_indices", []) or [])
                failed_indices = set(ckpt.get("failed_indices", []) or [])
                logger.info(
                    f"加载断点: 已处理 {len(processed_indices)}/{len(df)}，失败 {len(failed_indices)}"
                )
            except Exception as exc:
                logger.warning(f"加载断点失败，将从头开始。err={exc}")

        def save_checkpoint() -> None:
            """
            保存当前处理进度到断点文件
            包含总数、已处理索引列表、失败索引列表和时间戳
            """
            payload = {
                "total": int(len(df)),
                "processed_indices": sorted(int(i) for i in processed_indices),
                "failed_indices": sorted(int(i) for i in failed_indices),
                "timestamp": datetime.now().isoformat(timespec="seconds"),
            }
            with open(checkpoint_file, "w", encoding="utf-8") as f:
                json.dump(payload, f, ensure_ascii=False, indent=2)

        # 初始化进度条（若无tqdm则降级为普通迭代）
        iterator = df.iterrows()
        try:
            from tqdm import tqdm  # type: ignore
            iterator = tqdm(iterator, total=len(df), desc="Processing queries")
        except Exception:
            pass

        # 批量处理查询
        try:
            for idx, row in iterator:
                # 跳过已处理的查询
                if idx in processed_indices:
                    continue
                
                # 获取查询内容
                query = str(row.get("query", "")).strip()
                if not query:
                    # 空查询直接标记为已处理
                    processed_indices.add(idx)
                    save_checkpoint()
                    continue
                
                logger.info(f"开始处理第 {idx} 行 query: {query}")
                
                # 执行查询处理
                try:
                    self.signalparse(query)
                    processed_indices.add(idx)
                except Exception as exc:
                    # 记录失败的查询
                    failed_indices.add(idx)
                    logger.exception(f"处理 query 失败（行 {idx}）: {exc}")
                finally:
                    # 每步都保存断点，确保可断点续跑
                    save_checkpoint()
                    
        except KeyboardInterrupt:
            # 处理用户中断（Ctrl+C）
            logger.warning("检测到中断信号，保存进度后退出...")
            save_checkpoint()
    
    def batch_debug_queries(self, 
                           input_file: str = "data/input/queries.xlsx",
                           checkpoint_dir: str = "data/process/debug_checkpoints") -> None:
        """
        批量调试查询文件中的所有查询语句（从步骤2开始）
        
        该方法从Excel文件中读取查询列表，逐个调用debug_main_from_step2进行调试分析。
        用于定位空结果问题，绕过LLM调用，直接从已保存的LLM结果开始执行。
        
        Args:
            input_file (str): 输入的Excel文件路径，默认为"data/input/queries.xlsx"
            checkpoint_dir (str): 断点保存目录，默认为"data/process/debug_checkpoints"
        
        Returns:
            None
        
        功能说明:
            1. 从Excel文件中读取query列
            2. 支持断点保存，记录已处理和失败的查询索引
            3. 对每个查询调用self.debug()进行从步骤2开始的调试
            4. 使用tqdm显示进度条（如果可用）
            5. 捕获并记录处理过程中的异常
            6. 支持Ctrl+C中断，并保存当前进度
        
        异常处理:
            - 读取文件失败：记录错误日志并抛出异常
            - 单个查询调试失败：记录到failed_indices，继续处理下一个
            - 键盘中断：保存进度后优雅退出
        """
        # 读取输入文件
        try:
            df = DataReader.read_excel(input_file, columns=["query"])
        except Exception as exc:
            logger.error(f"读取 {input_file} 失败: {exc}")
            raise

        # 初始化断点保存设置
        os.makedirs(checkpoint_dir, exist_ok=True)
        checkpoint_file = os.path.join(checkpoint_dir, "debug_queries.progress.json")

        # 初始化处理状态集合
        processed_indices: set[int] = set()  # 已成功调试的查询索引
        failed_indices: set[int] = set()      # 调试失败的查询索引

        # 读取已有进度（断点恢复）
        if os.path.exists(checkpoint_file):
            try:
                with open(checkpoint_file, "r", encoding="utf-8") as f:
                    ckpt = json.load(f) or {}
                processed_indices = set(ckpt.get("processed_indices", []) or [])
                failed_indices = set(ckpt.get("failed_indices", []) or [])
                logger.info(
                    f"加载调试断点: 已处理 {len(processed_indices)}/{len(df)}，失败 {len(failed_indices)}"
                )
            except Exception as exc:
                logger.warning(f"加载调试断点失败，将从头开始。err={exc}")

        def save_checkpoint() -> None:
            """
            保存当前调试进度到断点文件
            包含总数、已处理索引列表、失败索引列表和时间戳
            """
            payload = {
                "total": int(len(df)),
                "processed_indices": sorted(int(i) for i in processed_indices),
                "failed_indices": sorted(int(i) for i in failed_indices),
                "timestamp": datetime.now().isoformat(timespec="seconds"),
            }
            with open(checkpoint_file, "w", encoding="utf-8") as f:
                json.dump(payload, f, ensure_ascii=False, indent=2)

        # 初始化进度条（若无tqdm则降级为普通迭代）
        iterator = df.iterrows()
        try:
            from tqdm import tqdm  # type: ignore
            iterator = tqdm(iterator, total=len(df), desc="Debugging queries")
        except Exception:
            pass

        # 批量调试查询
        try:
            for idx, row in iterator:
                # 跳过已处理的查询
                if idx in processed_indices:
                    continue
                
                # 获取查询内容
                query = str(row.get("query", "")).strip()
                if not query:
                    # 空查询直接标记为已处理
                    processed_indices.add(idx)
                    save_checkpoint()
                    continue
                
                logger.info(f"开始调试第 {idx} 行 query: {query}")
                
                # 执行调试（从步骤2开始）
                try:
                    self.debug(query)
                    processed_indices.add(idx)
                except FileNotFoundError as exc:
                    # LLM结果文件不存在，记录为失败
                    failed_indices.add(idx)
                    logger.error(f"调试 query 失败（行 {idx}）- LLM结果文件不存在: {exc}")
                except Exception as exc:
                    # 其他异常，记录失败
                    failed_indices.add(idx)
                    logger.exception(f"调试 query 失败（行 {idx}）: {exc}")
                finally:
                    # 每步都保存断点，确保可断点续跑
                    save_checkpoint()
                    
        except KeyboardInterrupt:
            # 处理用户中断（Ctrl+C）
            logger.warning("检测到中断信号，保存调试进度后退出...")
            save_checkpoint()
        
        # 输出最终统计信息
        logger.info(f"批量调试完成: 成功 {len(processed_indices)}/{len(df)}，失败 {len(failed_indices)}")
    
    def batch_generate_sql(self, 
                          input_file: str = "data/input/queries.xlsx",
                          retry_times: int = 5,
                          checkpoint_dir: str = "data/process/sql_generation_checkpoints",
                          use_signal_files: bool = True) -> None:
        """
        批量为查询生成 SQL
        
        该方法从Excel文件中读取查询列表，逐个调用SQLGenerator生成SQL。
        支持断点续跑功能，即使中途中断也可以从上次停止的地方继续执行。
        
        Args:
            input_file (str): 输入的Excel文件路径，默认为"data/input/queries.xlsx"
            retry_times (int): 每个查询调用LLM的次数，默认为5次
            checkpoint_dir (str): 断点保存目录，默认为"data/process/sql_generation_checkpoints"
            use_signal_files (bool): 是否使用信号文件，默认为True
                - True: 从 signal_parse 结果文件中自动加载信号（需要先运行 signalparse）
                - False: 不使用信号文件，LLM 将基于示例和上下文生成 SQL
        
        Returns:
            None
        
        功能说明:
            1. 从Excel文件中读取query列
            2. 支持断点保存，记录已处理和失败的查询索引
            3. 对每个查询调用SQLGenerator.generate_sql_by_CoT()生成SQL
            4. 使用tqdm显示进度条（如果可用）
            5. 捕获并记录处理过程中的异常
            6. 支持Ctrl+C中断，并保存当前进度
            7. 所有生成的SQL自动保存到配置文件指定的CSV路径
        
        异常处理:
            - 读取文件失败：记录错误日志并抛出异常
            - 单个查询生成失败：记录到failed_indices，继续处理下一个
            - 键盘中断：保存进度后优雅退出
        """
        # 读取输入文件
        try:
            df = DataReader.read_excel(input_file, columns=["query"])
        except Exception as exc:
            logger.error(f"读取 {input_file} 失败: {exc}")
            raise

        # 初始化断点保存设置
        os.makedirs(checkpoint_dir, exist_ok=True)
        checkpoint_file = os.path.join(checkpoint_dir, "sql_generation.progress.json")

        # 初始化处理状态集合
        processed_indices: set[int] = set()  # 已成功处理的查询索引
        failed_indices: set[int] = set()      # 处理失败的查询索引

        # 读取已有进度（断点恢复）
        if os.path.exists(checkpoint_file):
            try:
                with open(checkpoint_file, "r", encoding="utf-8") as f:
                    ckpt = json.load(f) or {}
                processed_indices = set(ckpt.get("processed_indices", []) or [])
                failed_indices = set(ckpt.get("failed_indices", []) or [])
                logger.info(
                    f"加载SQL生成断点: 已处理 {len(processed_indices)}/{len(df)}，失败 {len(failed_indices)}"
                )
            except Exception as exc:
                logger.warning(f"加载SQL生成断点失败，将从头开始。err={exc}")

        def save_checkpoint() -> None:
            """
            保存当前处理进度到断点文件
            包含总数、已处理索引列表、失败索引列表和时间戳
            """
            payload = {
                "total": int(len(df)),
                "processed_indices": sorted(int(i) for i in processed_indices),
                "failed_indices": sorted(int(i) for i in failed_indices),
                "timestamp": datetime.now().isoformat(timespec="seconds"),
            }
            with open(checkpoint_file, "w", encoding="utf-8") as f:
                json.dump(payload, f, ensure_ascii=False, indent=2)

        # 初始化进度条（若无tqdm则降级为普通迭代）
        iterator = df.iterrows()
        try:
            from tqdm import tqdm  # type: ignore
            iterator = tqdm(iterator, total=len(df), desc="Generating SQL")
        except Exception:
            pass

        # 批量生成SQL
        try:
            for idx, row in iterator:
                # 跳过已处理的查询
                if idx in processed_indices:
                    continue
                
                # 获取查询内容
                query = str(row.get("query", "")).strip()
                if not query:
                    # 空查询直接标记为已处理
                    processed_indices.add(idx)
                    save_checkpoint()
                    continue
                
                logger.info(f"开始为第 {idx} 行 query 生成 SQL: {query}")
                
                # 执行SQL生成
                try:
                    if use_signal_files:
                        # 使用信号文件（自动加载）
                        results = self.sqlgenerator.generate_sql_by_CoT(
                            query=query,
                            retry_times=retry_times
                        )
                    else:
                        # 不使用信号文件
                        results = self.sqlgenerator.generate_sql_by_CoT(
                            query=query,
                            retry_times=retry_times,
                            signal_info=[]  # 空信号列表
                        )
                    
                    # 记录生成结果统计
                    success_count = sum(1 for r in results if "error" not in r)
                    logger.info(
                        f"第 {idx} 行 SQL 生成完成: 成功 {success_count}/{len(results)} 次"
                    )
                    
                    # 检查信息充分性
                    is_sufficient = SQLGenerator.check_info_sufficiency(results)
                    if not is_sufficient:
                        logger.warning(f"第 {idx} 行: 信号信息可能不充分")
                    
                    processed_indices.add(idx)
                    
                except Exception as exc:
                    # 记录失败的查询
                    failed_indices.add(idx)
                    logger.exception(f"为 query 生成 SQL 失败（行 {idx}）: {exc}")
                finally:
                    # 每步都保存断点，确保可断点续跑
                    save_checkpoint()
                    
        except KeyboardInterrupt:
            # 处理用户中断（Ctrl+C）
            logger.warning("检测到中断信号，保存SQL生成进度后退出...")
            save_checkpoint()
        
        # 输出最终统计信息
        logger.info(f"批量SQL生成完成: 成功 {len(processed_indices)}/{len(df)}，失败 {len(failed_indices)}")
        
        # 提示结果保存位置
        sql_csv_path = self.sqlgenerator.config.get("sql_generate", {}).get(
            "SQL_Generate_With_Decomposation_By_CoT", {}
        ).get("sql_csv_path", "data/output/sql_generation_with_decomposition_by_cot.csv")
        logger.info(f"所有生成的SQL已保存到: {sql_csv_path}")
        print(f"\n{'='*80}")
        print(f"批量SQL生成完成！")
        print(f"成功: {len(processed_indices)}/{len(df)}")
        print(f"失败: {len(failed_indices)}/{len(df)}")
        print(f"结果保存位置: {sql_csv_path}")
        print(f"{'='*80}\n")

    def batch_iterative_sql_generation(self,
                                       input_file: str = "data/input/queries.xlsx",
                                       retry_times: int = 5,
                                       max_iterations: int = 3,
                                       top_k_additional: int = 6,
                                       checkpoint_dir: str = "data/process/iterative_sql_checkpoints") -> None:
        """
        批量迭代生成 SQL（自动补充信号信息）
        
        该方法使用 iterative_sql_generation 进行批量处理，会自动：
        1. 检查信息充分性
        2. 对信息不足的查询自动补充信号
        3. 迭代优化直到信息充足或达到最大迭代次数
        
        Args:
            input_file (str): 输入的Excel文件路径，默认为"data/input/queries.xlsx"
            retry_times (int): 每次迭代中调用LLM的次数，默认为5次
            max_iterations (int): 最大迭代次数，默认为3次
            top_k_additional (int): 搜索额外信号时的top_k，默认为10
            checkpoint_dir (str): 断点保存目录
            
        Returns:
            None
            
        功能说明:
            相比 batch_generate_sql，本方法会智能地补充缺失的信号信息，
            提高SQL生成的准确性和完整性。
        """
        # 读取输入文件
        try:
            df = DataReader.read_excel(input_file, columns=["query"])
        except Exception as exc:
            logger.error(f"读取 {input_file} 失败: {exc}")
            raise

        # 初始化断点保存设置
        os.makedirs(checkpoint_dir, exist_ok=True)
        checkpoint_file = os.path.join(checkpoint_dir, "iterative_sql_generation.progress.json")

        # 初始化处理状态集合
        processed_indices: set[int] = set()
        failed_indices: set[int] = set()

        # 加载断点（如果存在）
        def load_checkpoint():
            if os.path.exists(checkpoint_file):
                try:
                    with open(checkpoint_file, "r", encoding="utf-8") as f:
                        checkpoint_data = json.load(f)
                    processed_indices.update(checkpoint_data.get("processed", []))
                    failed_indices.update(checkpoint_data.get("failed", []))
                    logger.info(
                        f"从断点恢复: 已处理 {len(processed_indices)} 个，失败 {len(failed_indices)} 个"
                    )
                except Exception as e:
                    logger.warning(f"加载断点失败: {e}，将从头开始")

        # 保存断点
        def save_checkpoint():
            try:
                checkpoint_data = {
                    "processed": list(processed_indices),
                    "failed": list(failed_indices),
                    "total": len(df),
                    "timestamp": datetime.now().isoformat()
                }
                with open(checkpoint_file, "w", encoding="utf-8") as f:
                    json.dump(checkpoint_data, f, indent=2, ensure_ascii=False)
            except Exception as e:
                logger.error(f"保存断点失败: {e}")

        # 加载之前的进度
        load_checkpoint()

        # 显示进度信息
        logger.info(f"开始批量迭代SQL生成，共 {len(df)} 条查询")
        logger.info(f"参数: retry_times={retry_times}, max_iterations={max_iterations}, top_k={top_k_additional}")

        try:
            # 遍历所有查询
            for idx, row in df.iterrows():
                query = row["query"]
                
                # 跳过已失败的查询
                if idx in failed_indices:
                    logger.info(f"跳过之前失败的 query（行 {idx}）")
                    continue
                
                # 检查已处理的查询是否真正完成了所有迭代
                if idx in processed_indices:
                    # 统计该查询实际完成的迭代次数
                    completed_iterations = self._count_completed_iterations(query, max_iterations)
                    
                    if completed_iterations >= max_iterations:
                        logger.info(
                            f"跳过已完全处理的 query（行 {idx}，完成 {completed_iterations}/{max_iterations} 次迭代）"
                        )
                        continue
                    else:
                        logger.warning(
                            f"查询 {idx} 只完成了 {completed_iterations}/{max_iterations} 次迭代，将继续处理"
                        )
                        # 不跳过，让 iterative_sql_generation 从断点继续
                
                logger.info(f"\n{'='*80}")
                logger.info(f"处理第 {idx}/{len(df)-1} 行查询")
                logger.info(f"查询: {query[:100]}...")
                logger.info(f"{'='*80}")
                
                # 执行迭代SQL生成
                try:
                    results = self.sqlgenerator.iterative_sql_generation(
                        query=query,
                        retry_times=retry_times,
                        max_iterations=max_iterations,
                        top_k_additional=top_k_additional
                    )
                    
                    # 记录生成结果统计
                    success_count = sum(1 for r in results if "error" not in r)
                    is_sufficient = SQLGenerator.check_info_sufficiency(results)
                    
                    logger.info(
                        f"第 {idx} 行完成: 成功 {success_count}/{len(results)} 次, "
                        f"信息充分性: {'✅' if is_sufficient else '⚠️'}"
                    )
                    
                    processed_indices.add(idx)
                    
                except Exception as exc:
                    # 记录失败的查询
                    failed_indices.add(idx)
                    logger.exception(f"迭代生成SQL失败（行 {idx}）: {exc}")
                finally:
                    # 每步都保存断点
                    save_checkpoint()
                    
        except KeyboardInterrupt:
            # 处理用户中断（Ctrl+C）
            logger.warning("检测到中断信号，保存进度后退出...")
            save_checkpoint()
        
        # 输出最终统计信息
        logger.info(f"\n{'='*80}")
        logger.info(f"批量迭代SQL生成完成!")
        logger.info(f"成功: {len(processed_indices)}/{len(df)}")
        logger.info(f"失败: {len(failed_indices)}/{len(df)}")
        logger.info(f"{'='*80}")
        
        # 提示结果保存位置
        sql_csv_path = self.sqlgenerator.config.get("sql_generate", {}).get(
            "SQL_Generate_With_Decomposation_By_CoT", {}
        ).get("sql_csv_path", "data/output/sql_generation_with_decomposition_by_cot.csv")
        
        print(f"\n{'='*80}")
        print(f"批量迭代SQL生成完成！")
        print(f"成功: {len(processed_indices)}/{len(df)}")
        print(f"失败: {len(failed_indices)}/{len(df)}")
        print(f"结果保存位置: {sql_csv_path}")
        print(f"{'='*80}\n")

    def _count_completed_iterations(self, query: str, max_iterations: int) -> int:
        """
        统计某个查询已完成的迭代次数
        
        该方法会检查所有迭代文件，确定某个查询实际完成了多少次迭代。
        用于断点续跑时判断是否需要继续处理该查询。
        
        Args:
            query (str): 查询内容
            max_iterations (int): 最大迭代次数
            
        Returns:
            int: 已完成的迭代次数（0 到 max_iterations）
        
        示例:
            如果查询在 iter1.csv 和 iter2.csv 中都存在，但 iter3.csv 中不存在，
            则返回 2（完成了2次迭代）
        """
        import pandas as pd
        
        # 获取基础 CSV 路径
        sql_csv_path = self.sqlgenerator.config.get("sql_generate", {}).get(
            "SQL_Generate_With_Decomposation_By_CoT", {}
        ).get("sql_csv_path", "data/output/sql_generation_with_decomposition_by_cot.csv")
        
        base_path = os.path.splitext(sql_csv_path)[0]
        extension = os.path.splitext(sql_csv_path)[1]
        
        completed = 0
        
        # 检查每个迭代文件
        for i in range(1, max_iterations + 1):
            # 生成文件路径
            if i == 1:
                file_path = sql_csv_path
            else:
                file_path = f"{base_path}_iter{i}{extension}"
            
            # 检查文件是否存在且包含该查询
            if os.path.exists(file_path):
                try:
                    df = pd.read_csv(file_path)
                    if "query" in df.columns and query in df["query"].values:
                        completed = i  # 更新完成的迭代次数
                    else:
                        # 该查询不在当前迭代文件中，停止检查
                        break
                except Exception as e:
                    logger.warning(f"读取迭代文件 {file_path} 失败: {e}")
                    break
            else:
                # 文件不存在，停止检查
                break
        
        return completed

    def analyze_iterative_sql_statistics(self, max_iterations: int = 3) -> dict:
        """
        分析迭代 SQL 生成结果的统计信息
        
        该方法会读取所有迭代版本的 SQL 生成结果文件，统计每个查询在各次迭代中
        is_enough_info 字段的变化模式，例如：
        - 一直为 True (TTT)
        - 第一次 False 后来 True (FTT)
        - 所有组合 (FFF, FFT, FTF, FTT, TFF, TFT, TTF, TTT)
        
        Args:
            max_iterations (int): 最大迭代次数，默认为3次
            
        Returns:
            dict: 包含统计结果的字典，格式为：
                {
                    "total_queries": int,  # 总查询数
                    "patterns": {          # 各种模式的统计
                        "TTT": int,        # 一直为 True 的数量
                        "FTT": int,        # False -> True -> True
                        ...
                    },
                    "pattern_details": {   # 每个模式包含的查询列表
                        "TTT": [query1, query2, ...],
                        ...
                    },
                    "iteration_files": [   # 使用的文件列表
                        "file1.csv",
                        "file2.csv",
                        ...
                    ]
                }
        """
        import pandas as pd
        from collections import defaultdict
        
        # 获取基础 CSV 路径
        sql_csv_path = self.sqlgenerator.config.get("sql_generate", {}).get(
            "SQL_Generate_With_Decomposation_By_CoT", {}
        ).get("sql_csv_path", "data/output/sql_generation_with_decomposition_by_cot.csv")
        
        base_path = os.path.splitext(sql_csv_path)[0]
        extension = os.path.splitext(sql_csv_path)[1]
        
        # 收集所有迭代文件
        iteration_files = []
        for i in range(1, max_iterations + 1):
            if i == 1:
                file_path = sql_csv_path
            else:
                file_path = f"{base_path}_iter{i}{extension}"
            
            if os.path.exists(file_path):
                iteration_files.append(file_path)
                logger.info(f"找到迭代文件 {i}: {file_path}")
            else:
                logger.warning(f"迭代文件 {i} 不存在: {file_path}")
        
        if not iteration_files:
            logger.error("未找到任何迭代文件！")
            return {
                "total_queries": 0,
                "patterns": {},
                "pattern_details": {},
                "iteration_files": []
            }
        
        # 读取所有迭代文件的数据
        iteration_data = []
        for file_path in iteration_files:
            try:
                df = pd.read_csv(file_path)
                logger.info(f"成功读取 {file_path}, 共 {len(df)} 条记录")
                iteration_data.append(df)
            except Exception as e:
                logger.error(f"读取文件失败 {file_path}: {e}")
                return {
                    "total_queries": 0,
                    "patterns": {},
                    "pattern_details": {},
                    "iteration_files": iteration_files
                }
        
        # 获取所有唯一的查询
        all_queries = set()
        for df in iteration_data:
            if "query" in df.columns:
                all_queries.update(df["query"].unique())
        
        logger.info(f"共找到 {len(all_queries)} 个唯一查询")
        
        # 统计每个查询的 is_enough_info 模式
        query_patterns = {}  # query -> pattern (如 "TTF")
        pattern_counts = defaultdict(int)  # pattern -> count
        pattern_queries = defaultdict(list)  # pattern -> [queries]
        
        for query in all_queries:
            pattern = []
            last_state = "N"  # 记录最后一个有效状态，用于状态继承
            
            for df in iteration_data:
                # 查找该查询在当前迭代中的记录
                query_rows = df[df["query"] == query] if "query" in df.columns else pd.DataFrame()
                
                if query_rows.empty:
                    # 如果该查询在当前迭代中不存在
                    # 检查是否应该继承上一次的状态（迭代提前终止）
                    if last_state == "T":
                        # 如果上一次是True（信息充分），说明迭代已终止，继承T状态
                        pattern.append("T")
                        logger.debug(f"查询继承状态: {query[:30]}... -> T (迭代已终止)")
                    else:
                        # 否则标记为N（真正不存在）
                        pattern.append("N")
                else:
                    # ⭐ 修复：使用多数投票判断状态，而不是只看第一条记录
                    # 统计所有记录中 is_enough_info 的 True/False 数量
                    true_count = 0
                    false_count = 0
                    
                    for _, row in query_rows.iterrows():
                        is_enough = row.get("is_enough_info", None)
                        if pd.notna(is_enough):
                            if is_enough in [True, "True", "true", 1, "1"]:
                                true_count += 1
                            else:
                                false_count += 1
                    
                    # 多数投票决定状态
                    if true_count == 0 and false_count == 0:
                        state = "N"
                    elif true_count > false_count:
                        state = "T"
                    else:
                        state = "F"
                    
                    pattern.append(state)
                    last_state = state  # 更新最后一个有效状态
            
            # 将模式转换为字符串
            pattern_str = "".join(pattern)
            query_patterns[query] = pattern_str
            pattern_counts[pattern_str] += 1
            pattern_queries[pattern_str].append(query)
        
        # 输出统计结果
        logger.info(f"\n{'='*80}")
        logger.info(f"迭代 SQL 生成统计结果")
        logger.info(f"{'='*80}")
        logger.info(f"总查询数: {len(all_queries)}")
        logger.info(f"迭代次数: {len(iteration_files)}")
        logger.info(f"\n模式说明: T=信息充分, F=信息不足, N=不存在（真正未处理）")
        logger.info(f"注意: 如果某查询在第N次达到T后，第N+1次显示的T是继承状态（迭代已终止）")
        logger.info(f"\n各种模式的统计:")
        
        # 按模式排序输出
        sorted_patterns = sorted(pattern_counts.items(), key=lambda x: x[1], reverse=True)
        
        print(f"\n{'='*80}")
        print(f"迭代 SQL 生成统计结果（使用状态继承逻辑）")
        print(f"{'='*80}")
        print(f"总查询数: {len(all_queries)}")
        print(f"迭代文件数: {len(iteration_files)}")
        print(f"\n模式说明: T=信息充分, F=信息不足, N=不存在（真正未处理）")
        print(f"注意: 当某查询达到T后提前终止，后续迭代显示的T是继承状态")
        print(f"\n{'模式':<10} {'数量':>8} {'占比':>8} {'示例查询'}")
        print(f"{'-'*80}")
        
        for pattern, count in sorted_patterns:
            percentage = (count / len(all_queries)) * 100
            example_query = pattern_queries[pattern][0][:40] if pattern_queries[pattern] else ""
            print(f"{pattern:<10} {count:>8} {percentage:>7.1f}% {example_query}...")
            logger.info(f"{pattern}: {count} ({percentage:.1f}%)")
        
        print(f"{'='*80}\n")
        
        # 特别统计一些关键模式
        key_patterns = {
            "一直充分 (全T)": [p for p in pattern_counts.keys() if all(c == "T" for c in p if c != "N")],
            "逐步改善 (F->T)": [p for p in pattern_counts.keys() if "F" in p and p.endswith("T") and "N" not in p],
            "一直不足 (全F)": [p for p in pattern_counts.keys() if all(c == "F" for c in p if c != "N")],
            "反复波动": [p for p in pattern_counts.keys() if "T" in p and "F" in p and p.count("T") > 1 and p.count("F") > 1],
        }
        
        print(f"关键模式分析:")
        print(f"{'-'*80}")
        for category, patterns in key_patterns.items():
            category_count = sum(pattern_counts[p] for p in patterns)
            if category_count > 0:
                percentage = (category_count / len(all_queries)) * 100
                print(f"{category:<20} {category_count:>8} ({percentage:>5.1f}%)")
                logger.info(f"{category}: {category_count} ({percentage:.1f}%)")
        
        print(f"{'='*80}\n")
        
        # 返回统计结果
        return {
            "total_queries": len(all_queries),
            "patterns": dict(pattern_counts),
            "pattern_details": dict(pattern_queries),
            "iteration_files": iteration_files,
            "key_patterns": {k: sum(pattern_counts[p] for p in v) for k, v in key_patterns.items()}
        }

    def single_iterative_sql_generation(self,
                                        input_file: str = "data/input/queries.xlsx",
                                        query_index: int = None,
                                        query_text: str = None,
                                        retry_times: int = 5,
                                        max_iterations: int = 3,
                                        top_k_additional: int = 6) -> list:
        """
        对 xlsx 文件中的某个特定查询执行迭代 SQL 生成
        
        可以通过索引或查询文本选择要处理的查询。
        
        Args:
            input_file (str): 输入的Excel文件路径，默认为"data/input/queries.xlsx"
            query_index (int): 查询在文件中的行索引（从0开始），优先使用
            query_text (str): 查询文本（部分匹配），当 query_index 为 None 时使用
            retry_times (int): 每次迭代中调用LLM的次数，默认为5次
            max_iterations (int): 最大迭代次数，默认为3次
            top_k_additional (int): 搜索额外信号时的top_k，默认为6
            
        Returns:
            list: 最后一次迭代的生成结果列表
            
        Raises:
            ValueError: 当未提供 query_index 或 query_text 时
            IndexError: 当 query_index 超出范围时
            ValueError: 当使用 query_text 未找到匹配的查询时
            
        使用示例:
            # 方法1: 通过索引选择（推荐）
            results = analyst.single_iterative_sql_generation(
                query_index=7,  # 处理第8行的查询（索引从0开始）
                retry_times=5,
                max_iterations=3
            )
            
            # 方法2: 通过文本匹配选择
            results = analyst.single_iterative_sql_generation(
                query_text="TPMS",  # 处理包含"TPMS"的查询
                retry_times=5,
                max_iterations=3
            )
        """
        # 验证参数
        if query_index is None and query_text is None:
            raise ValueError("必须提供 query_index 或 query_text 参数之一")
        
        # 读取输入文件
        try:
            df = DataReader.read_excel(input_file, columns=["query"])
        except Exception as exc:
            logger.error(f"读取 {input_file} 失败: {exc}")
            raise
        
        # 选择目标查询
        selected_query = None
        selected_index = None
        
        if query_index is not None:
            # 通过索引选择
            if query_index < 0 or query_index >= len(df):
                raise IndexError(
                    f"query_index {query_index} 超出范围（有效范围: 0-{len(df)-1}）"
                )
            selected_query = df.iloc[query_index]["query"]
            selected_index = query_index
            logger.info(f"通过索引选择查询: 第 {query_index} 行")
        else:
            # 通过文本匹配选择
            matching_rows = df[df["query"].str.contains(query_text, na=False, case=False)]
            
            if matching_rows.empty:
                raise ValueError(f"未找到包含 '{query_text}' 的查询")
            
            if len(matching_rows) > 1:
                logger.warning(
                    f"找到 {len(matching_rows)} 个匹配的查询，将使用第一个匹配项"
                )
                print(f"\n找到多个匹配的查询:")
                for idx, row in matching_rows.iterrows():
                    print(f"  索引 {idx}: {row['query'][:60]}...")
                print(f"\n将使用第一个匹配项（索引 {matching_rows.index[0]}）\n")
            
            selected_index = matching_rows.index[0]
            selected_query = matching_rows.iloc[0]["query"]
            logger.info(f"通过文本匹配选择查询: 第 {selected_index} 行")
        
        # 显示选中的查询信息
        print(f"\n{'='*80}")
        print(f"单个查询迭代 SQL 生成")
        print(f"{'='*80}")
        print(f"查询索引: {selected_index}")
        print(f"查询内容: {selected_query}")
        print(f"\n参数:")
        print(f"  - retry_times: {retry_times}")
        print(f"  - max_iterations: {max_iterations}")
        print(f"  - top_k_additional: {top_k_additional}")
        print(f"{'='*80}\n")
        
        logger.info(f"开始为查询 {selected_index} 生成迭代 SQL")
        logger.info(f"查询内容: {selected_query}")
        
        # 执行迭代SQL生成
        try:
            results = self.sqlgenerator.iterative_sql_generation(
                query=selected_query,
                retry_times=retry_times,
                max_iterations=max_iterations,
                top_k_additional=top_k_additional
            )
            
            # 统计结果
            success_count = sum(1 for r in results if "error" not in r)
            is_sufficient = SQLGenerator.check_info_sufficiency(results)
            
            print(f"\n{'='*80}")
            print(f"生成完成！")
            print(f"{'='*80}")
            print(f"成功次数: {success_count}/{len(results)}")
            print(f"信息充分性: {'✅ 充分' if is_sufficient else '⚠️ 不充分'}")
            
            # 提示结果保存位置
            sql_csv_path = self.sqlgenerator.config.get("sql_generate", {}).get(
                "SQL_Generate_With_Decomposation_By_CoT", {}
            ).get("sql_csv_path", "data/output/sql_generation_with_decomposition_by_cot.csv")
            print(f"结果已保存到: {sql_csv_path}")
            print(f"{'='*80}\n")
            
            logger.info(
                f"查询 {selected_index} 完成: 成功 {success_count}/{len(results)} 次, "
                f"信息充分性: {'✅' if is_sufficient else '⚠️'}"
            )
            
            return results
            
        except Exception as exc:
            logger.exception(f"为查询 {selected_index} 生成 SQL 失败: {exc}")
            print(f"\n❌ 生成失败: {exc}\n")
            raise

    def batch_final_sql_vote(self,
                            input_file: str = "data/input/candidate_sqls.csv",
                            output_file: str = "data/output/final_sql.csv",
                            max_concurrent: int = 5) -> None:
        """
        批量评估候选SQL并选出最优SQL
        
        该方法从CSV文件中读取候选SQL列表，对于每个唯一的查询需求，
        异步并发调用LLM进行评估。
        
        Args:
            input_file (str): 输入CSV文件路径，必须包含以下列：
                - query: 自然语言业务需求
                - sql: 候选SQL语句
                - relevant_signals: (可选) 相关信号列表
            output_file (str): 输出CSV文件路径，包含以下列：
                - query: 原始业务需求
                - final_sql: 最终选择的SQL
                - relevant_signals: 相关信号列表（从输入复制）
                - evaluation_score: 评分
                - selection_reason: 选择理由
            max_concurrent (int): 最大并发LLM调用数，默认为5
        
        Returns:
            None
        
        功能说明:
            1. 从CSV文件中读取候选SQL
            2. 按query分组，每个query对应多个候选SQL
            3. 使用异步并发方式同时处理多个查询（最多同时处理max_concurrent个）
            4. 为每个查询选出最优SQL
            5. 使用tqdm显示进度条
        """
        import pandas as pd
        from tqdm import tqdm
        import asyncio
        
        # 读取输入文件
        try:
            df = pd.read_csv(input_file)
            logger.info(f"成功读取输入文件: {input_file}, 共 {len(df)} 行")
        except Exception as exc:
            logger.error(f"读取 {input_file} 失败: {exc}")
            raise
        
        # 验证必需列
        required_columns = ['query', 'sql']
        missing_columns = [col for col in required_columns if col not in df.columns]
        if missing_columns:
            raise ValueError(f"输入文件缺少必需列: {missing_columns}")
        
        # 按query分组
        grouped = df.groupby('query', sort=False)
        total_queries = len(grouped)
        
        logger.info(f"共有 {total_queries} 个唯一查询需求需要处理")
        
        # 存储结果
        results = []
        
        # 定义异步处理单个查询的函数
        async def process_query_async(query: str, candidate_sqls: list, relevant_signals: str):
            """
            异步处理单个查询，调用LLM进行评估
            
            Args:
                query: 查询需求
                candidate_sqls: 候选SQL列表
                relevant_signals: 相关信号
            
            Returns:
                评估结果字典
            """
            try:
                logger.info(f"开始评估查询: {query[:100]}...")
                logger.info(f"该查询有 {len(candidate_sqls)} 个候选SQL")
                
                # 异步调用LLM一次
                response = await self.llm_client.async_llm_final_sql_vote(query, candidate_sqls)
                
                # 解析响应
                if isinstance(response, dict):
                    final_decision = response.get('final_decision', {})
                    selected_sql = final_decision.get('selected_sql', '')
                    total_score = final_decision.get('total_score', 0)
                    selection_reason = final_decision.get('selection_reason', '')
                elif isinstance(response, str):
                    # 如果返回的是字符串，尝试解析JSON
                    try:
                        import json
                        response_dict = json.loads(response)
                        final_decision = response_dict.get('final_decision', {})
                        selected_sql = final_decision.get('selected_sql', '')
                        total_score = final_decision.get('total_score', 0)
                        selection_reason = final_decision.get('selection_reason', '')
                    except json.JSONDecodeError:
                        logger.warning(f"无法解析LLM响应为JSON")
                        selected_sql = response
                        total_score = 0
                        selection_reason = "无法解析评估详情"
                else:
                    logger.warning(f"未知的响应格式: {type(response)}")
                    selected_sql = str(response)
                    total_score = 0
                    selection_reason = "未知响应格式"
                
                logger.info(f"查询评估完成，得分: {total_score}")
                
                return {
                    'query': query,
                    'final_sql': selected_sql,
                    'relevant_signals': relevant_signals,
                    'evaluation_score': total_score,
                    'selection_reason': selection_reason
                }
                
            except Exception as exc:
                logger.exception(f"评估查询失败: {exc}")
                return {
                    'query': query,
                    'final_sql': '',
                    'relevant_signals': relevant_signals,
                    'evaluation_score': 0,
                    'selection_reason': f"评估失败: {str(exc)}"
                }
        
        # 主处理逻辑 - 使用信号量控制并发数
        async def process_all_queries():
            """处理所有查询的异步函数，使用信号量限制并发"""
            # 创建信号量来限制并发数
            semaphore = asyncio.Semaphore(max_concurrent)
            
            async def process_with_semaphore(query, candidate_sqls, relevant_signals):
                """使用信号量控制的处理函数"""
                async with semaphore:
                    return await process_query_async(query, candidate_sqls, relevant_signals)
            
            # 为每个查询组创建异步任务
            tasks = []
            for query, group in grouped:
                candidate_sqls = group['sql'].tolist()
                relevant_signals = group['relevant_signals'].iloc[0] if 'relevant_signals' in group.columns else ""
                
                task = process_with_semaphore(query, candidate_sqls, relevant_signals)
                tasks.append(task)
            
            # 使用tqdm显示进度
            results_list = []
            with tqdm(total=len(tasks), desc=f"评估SQL (并发数: {max_concurrent})") as pbar:
                for coro in asyncio.as_completed(tasks):
                    result = await coro
                    results_list.append(result)
                    pbar.update(1)
            
            return results_list
        
        # 处理每个查询组（使用asyncio运行异步任务）
        try:
            # 运行异步处理
            results = asyncio.run(process_all_queries())
        except Exception as exc:
            logger.exception(f"异步处理失败: {exc}")
            raise
        
        # 保存结果
        try:
            # 确保输出目录存在
            output_dir = os.path.dirname(output_file)
            if output_dir:
                os.makedirs(output_dir, exist_ok=True)
            
            # 转换为DataFrame并保存
            results_df = pd.DataFrame(results)
            results_df.to_csv(output_file, index=False, encoding='utf-8-sig')
            
            logger.info(f"结果已保存到: {output_file}")
            logger.info(f"成功处理 {len(results)} 个查询")
            
            # 打印统计信息
            successful = sum(1 for r in results if r['final_sql'])
            print(f"\n{'='*80}")
            print(f"批量SQL评估完成！")
            print(f"{'='*80}")
            print(f"总查询数: {total_queries}")
            print(f"成功评估: {successful}")
            print(f"失败: {total_queries - successful}")
            print(f"结果保存至: {output_file}")
            print(f"{'='*80}\n")
            
        except Exception as exc:
            logger.error(f"保存结果失败: {exc}")
            raise

#TODO: 一个统一可配置的内部数据结构，比如当前用的是neo4j


# if __name__ == "__main__":
#     # 实例化LLMAsAnalyst
#     llm_analyst = LLMAsAnalyst()
    
#     # 批量处理查询
#     llm_analyst.batch_process_queries()