第三章：PandasQueryEngine - 小数据量高效方案

在处理小于1万行的结构化数据时，PandasQueryEngine 是最直接高效的解决方案：

• ✅ 零索引成本: 无需向量化，直接操作内存中的 DataFrame
• ✅ 精准计算: LLM 生成 Pandas 代码，计算100%准确
• ✅ 快速迭代: 适合探索性数据分析
• ⚠️ 适用场景: 数据量 < 1万行，单表或少量表

3.1 入门：基础查询

环境准备与数据加载

import pandas as pd
import time
from typing import Dict, Any, List
from llama_index.experimental.query_engine import PandasQueryEngine
from llama_index.llms.openai import OpenAI
from llama_index.core import Settings

import logging
# 禁用 Phoenix/OpenInference 的错误日志，在后续有提到的监控工具
logging.getLogger("openinference.instrumentation.llama_index").setLevel(logging.CRITICAL)

# 加载产品数据
df = pd.read_csv("dataset/products.csv")
print(f"📊 数据加载成功: {len(df)} 行 x {len(df.columns)} 列")
print(f"\n列名: {list(df.columns)}")
print("\n数据预览:")
df.head()

📊 数据加载成功: 32 行 x 9 列

列名: ['product_id', 'name', 'description', 'tags', 'price', 'category', 'stock', 'brand', 'rating']

数据预览:

<div> <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; }

.dataframe tbody tr th { vertical-align: top; }

.dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>product_id</th> <th>name</th> <th>description</th> <th>tags</th> <th>price</th> <th>category</th> <th>stock</th> <th>brand</th> <th>rating</th> </tr> </thead> <tbody> <tr> <th>0</th> <td>1</td> <td>MacBook Pro 14英寸</td> <td>MacBook Pro 14英寸是一款旗舰级的笔记本产品，续航持久，游戏体验佳。</td> <td>轻薄, 办公, 游戏, 高性能</td> <td>15087</td> <td>笔记本</td> <td>25</td> <td>苹果</td> <td>3.6</td> </tr> <tr> <th>1</th> <td>2</td> <td>ThinkPad X1 Carbon</td> <td>ThinkPad X1 Carbon采用机械轴体技术，配置强大，使用便捷，适合设计师使用。</td> <td>轻薄, 设计, 办公</td> <td>7611</td> <td>笔记本</td> <td>398</td> <td>联想</td> <td>4.7</td> </tr> <tr> <th>2</th> <td>3</td> <td>戴尔XPS 13</td> <td>这款戴尔XPS 13设计精美，性价比高且游戏体验佳，是笔记本中的明星产品。</td> <td>设计, 高性能</td> <td>10224</td> <td>笔记本</td> <td>319</td> <td>华硕</td> <td>3.9</td> </tr> <tr> <th>3</th> <td>4</td> <td>华硕灵耀14</td> <td>华硕灵耀14采用最新芯片技术，音质优秀，办公效率高，适合程序员使用。</td> <td>轻薄, 设计, 编程, 高性能</td> <td>8803</td> <td>笔记本</td> <td>45</td> <td>苹果</td> <td>3.6</td> </tr> <tr> <th>4</th> <td>5</td> <td>联想小新Pro 14</td> <td>这款联想小新Pro 14设计精美，高性能且办公效率高，是笔记本中的经典之作。</td> <td>编程, 设计, 轻薄</td> <td>14415</td> <td>笔记本</td> <td>117</td> <td>联想</td> <td>4.5</td> </tr> </tbody> </table> </div>

创建 PandasQueryEngine

PandasQueryEngine 的核心参数:

参数	说明
df	要查询的 DataFrame
verbose	是否显示生成的 Pandas 代码
instruction_str	指导 LLM 如何生成代码
synthesize_response	是否用 LLM 综合最终响应

# 优化的指令字符串 - 关键配置
instruction_str = (
    "1. 将查询转换为可执行的 Pandas 代码\n"
    "2. 代码必须是单行表达式，可以被 eval() 执行\n"
    "3. DataFrame 变量名为 'df'\n"
    "4. 只输出表达式本身，不要有其他文字\n"
    "5. 不要使用引号包裹表达式\n"
    "\n示例:\n"
    "问题: 找出价格最高的产品\n"
    "代码: df.loc[df['price'].idxmax()]['name']\n"
)

# 创建查询引擎
query_engine = PandasQueryEngine(
    df=df,   # DataFrame读取的数据
    verbose=True,  # 显示生成的代码
    instruction_str=instruction_str,
    synthesize_response=True  # 综合回答
)

print("✅ PandasQueryEngine 初始化完成")

✅ PandasQueryEngine 初始化完成

# 场景 1: 简单聚合查询
query = "电子产品类别有多少个产品？"
print(f"📝 查询: {query}")

response = query_engine.query(query)
print(f"💬 回答: {response}")

📝 查询: 电子产品类别有多少个产品？

Pandas Instructions:

df['category'].nunique()

Pandas Output: 6 💬 回答: 根据查询结果，电子产品类别中共有6种不同的产品。

# 场景 2: 排序和筛选
query = "列出价格(price)最高的5个产品，显示名称和价格"
print(f"\n📝 查询: {query}")

response = query_engine.query(query)
print(f"💬 回答: {response}")

📝 查询: 列出价格(price)最高的5个产品，显示名称和价格

Pandas Instructions:

df[['name', 'price']].sort_values(by='price', ascending=False).head(5)

Pandas Output: name price 0 MacBook Pro 14英寸 15087 4 联想小新Pro 14 14415 2 戴尔XPS 13 10224 5 惠普战66 10216 6 iPhone 15 Pro 9480 💬 回答: 根据查询结果，价格最高的5个产品及其价格如下：

1. MacBook Pro 14英寸 - 价格：15087元
2. 联想小新Pro 14 - 价格：14415元
3. 戴尔XPS 13 - 价格：10224元
4. 惠普战66 - 价格：10216元
5. iPhone 15 Pro - 价格：9480元

# 展示Pandas查询结果
df[['name', 'price']].sort_values(by='price', ascending=False).head(5)

<div> <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; }

.dataframe tbody tr th { vertical-align: top; }

.dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>name</th> <th>price</th> </tr> </thead> <tbody> <tr> <th>0</th> <td>MacBook Pro 14英寸</td> <td>15087</td> </tr> <tr> <th>4</th> <td>联想小新Pro 14</td> <td>14415</td> </tr> <tr> <th>2</th> <td>戴尔XPS 13</td> <td>10224</td> </tr> <tr> <th>5</th> <td>惠普战66</td> <td>10216</td> </tr> <tr> <th>6</th> <td>iPhone 15 Pro</td> <td>9480</td> </tr> </tbody> </table> </div>

3.2 进阶：复杂查询与性能监控

在生产环境中需要:

• 🔄 错误重试: 处理 LLM 偶发错误
• ⏱️ 性能监控: 跟踪查询耗时
• 📊 统计分析: 了解查询成功率

class AdvancedPandasQueryEngine:
    """增强的 Pandas 查询引擎 - 带错误处理和性能监控"""
    
    def __init__(self, df: pd.DataFrame, llm_model: str = "gpt-4o-mini"):
        self.df = df
        self.llm = OpenAI(model=llm_model, temperature=0)
        
        # 优化的指令（中文提示效果更好）
        instruction_str = (
            "你是数据分析专家，将自然语言转换为 Pandas 代码。\n"
            "要求:\n"
            "1. 生成单行可 eval() 执行的表达式\n"
            "2. DataFrame 变量名为 'df'\n"
            "3. 只输出代码，无其他文字\n"
            "4. 优先使用向量化操作\n"
        )
        
        self.query_engine = PandasQueryEngine(
            df=df, verbose=True,
            instruction_str=instruction_str,
            synthesize_response=True
        )
        self.query_history = []
    
    def query(self, query_str: str, max_retries: int = 2) -> Dict[str, Any]:
        """执行查询，带重试和性能记录"""
        start_time = time.time()
        
        for attempt in range(max_retries + 1):
            try:
                logger.info(f"执行查询 (尝试 {attempt + 1}): {query_str}")
                response = self.query_engine.query(query_str)
                execution_time = time.time() - start_time
                
                self.query_history.append({
                    "query": query_str,
                    "time": execution_time,
                    "success": True
                \})
                
                return {"success": True, "result": str(response), "time": execution_time}
                
            except Exception as e:
                if attempt == max_retries:
                    self.query_history.append({
                        "query": query_str,
                        "time": time.time() - start_time,
                        "success": False,
                        "error": str(e)
                    \})
                    return {"success": False, "error": str(e)}
    
    def get_stats(self) -> Dict[str, Any]:
        """获取性能统计"""
        if not self.query_history:
            return {"message": "没有查询历史"}
        
        successful = [q for q in self.query_history if q["success"]]
        return {
            "total": len(self.query_history),
            "success": len(successful),
            "success_rate": f"{len(successful)/len(self.query_history)*100:.1f}%",
            "avg_time": f"{sum(q['time'] for q in successful)/len(successful):.2f}s" if successful else "N/A"
        \}

# 初始化高级引擎
advanced_engine = AdvancedPandasQueryEngine(df)
print("✅ AdvancedPandasQueryEngine 初始化完成")

✅ AdvancedPandasQueryEngine 初始化完成

# 进阶查询示例
advanced_queries = [
    "计算每个品牌(brand)的产品数量，按数量降序排列",
    "找出评分(rating)高于4.5的产品名称和价格",
    "显示各类别(category)的平均价格"
]

print("=" * 60)
print("📊 进阶查询测试")
print("=" * 60)

for query in advanced_queries:
    print(f"\n📝 查询: {query}")
    result = advanced_engine.query(query)
    if result["success"]:
        print(f"💬 回答: {result['result']}")
        print(f"⏱️  耗时: {result['time']:.2f}s")
    else:
        print(f"❌ 错误: {result['error']}")

# 显示统计
print("\n" + "=" * 60)
print("📈 性能统计:")
stats = advanced_engine.get_stats()
for k, v in stats.items():
    print(f"   {k}: {v}")

============================================================ 📊 进阶查询测试

📝 查询: 计算每个品牌(brand)的产品数量，按数量降序排列 INFO:main:执行查询 (尝试 1): 计算每个品牌(brand)的产品数量，按数量降序排列 执行查询 (尝试 1): 计算每个品牌(brand)的产品数量，按数量降序排列

Pandas Instructions:

df['brand'].value_counts().sort_values(ascending=False)

Pandas Output: brand 苹果 11 三星 4 .. vivo 1 赛睿 1 Name: count, Length: 11, dtype: int64 💬 回答: 根据查询结果，各品牌的产品数量按降序排列如下：

• 苹果 (Apple): 11件
• 三星 (Samsung): 4件
• ...（此处省略了部分数据）
• vivo: 1件
• 赛睿 (SteelSeries): 1件

总共有11个不同的品牌被统计在内。从这个列表可以看出，“苹果”品牌的产品数量最多，达到了11件；而“vivo”和“赛睿”等品牌则各有1件产品。 ⏱️ 耗时: 5.41s

📝 查询: 找出评分(rating)高于4.5的产品名称和价格 INFO:main:执行查询 (尝试 1): 找出评分(rating)高于4.5的产品名称和价格 执行查询 (尝试 1): 找出评分(rating)高于4.5的产品名称和价格

Pandas Instructions:

df.loc[df['rating'] > 4.5, ['name', 'price']]

Pandas Output: name price 1 ThinkPad X1 Carbon 7611 7 华为Mate 60 4152 .. ... ... 29 雷蛇毒蝰 327 31 微软Arc鼠标 139

[11 rows x 2 columns] 💬 回答: 根据您的查询，以下是评分高于4.5的产品及其价格：

• ThinkPad X1 Carbon, 价格为7611元
• 华为Mate 60, 价格为4152元
• 雷蛇毒蝰, 价格为327元
• 微软Arc鼠标, 价格为139元

此外，还有其他几款产品也满足条件，但这里仅列出了部分示例。总共有11款产品的评分超过了4.5分。如果您需要完整列表或更多信息，请告诉我！ ⏱️ 耗时: 4.94s

📝 查询: 显示各类别(category)的平均价格 INFO:main:执行查询 (尝试 1): 显示各类别(category)的平均价格 执行查询 (尝试 1): 显示各类别(category)的平均价格

Pandas Instructions:

df.groupby('category')['price'].mean()

Pandas Output: category 平板 4569.0 手机 7767.0 ...
键盘 931.8 鼠标 408.4 Name: price, Length: 6, dtype: float64 💬 回答: 根据查询结果，各类别商品的平均价格如下：

• 平板: 4569.0元
• 手机: 7767.0元
• ...（此处省略了部分数据）
• 键盘: 931.8元
• 鼠标: 408.4元

这些数值代表了每个类别下所有商品价格的平均值。如果您需要更详细的信息或者其他类别的平均价格，请告诉我！ ⏱️ 耗时: 3.54s

============================================================ 📈 性能统计: total: 3 success: 3 success_rate: 100.0% avg_time: 4.63s

3.3 高阶：Chain of Tables（表格思维链）

普通的 Text-to-SQL（或 Text-to-Pandas）往往试图用一句超级复杂的 SQL/代码解决所有问题，这很容易出错（LLM 手滑写错代码）。对于复杂的多步骤分析，使用表格思维链技术:

原始表格 → [筛选] → 中间表1 → [聚合] → 中间表2 → [比较] → 最终结果

核心思想:

1. 分解复杂问题 为简单步骤
2. 逐步执行 每个操作
3. 保留中间结果 便于调试
4. 记录操作历史 可追溯

• 核心实现：

  • 这个实现是一个人机协同的 Chain-of-Table：人（或上层 Agent）：负责规划思维链（High-level reasoning），决定先 filter再 aggregate。

  • Python 类：负责维护中间状态（State Management），传递 current_df。

  • LLM：负责执行具体的原子操作（Atomic Execution），即把自然语言转成 Pandas 代码。

class ChainOfTablesEngine:
    """表格思维链引擎 - 多步骤推理（修正版）"""
    
    def __init__(self, df: pd.DataFrame):
        self.original_df = df.copy()    # 保存原始数据
        self.current_df = df.copy()    # 当前工作表
        self.operation_history = []    # 操作历史
        self.llm = Settings.llm        # LLM
    
    def filter(self, condition: str) -> pd.DataFrame:
        """筛选操作 - 真正更新 DataFrame"""
        print(f"[筛选] {condition}")
        rows_before = len(self.current_df)
        
        # 构造prompt，让LLM生成筛选代码
        prompt = f"""
        你是一位 Pandas 专家。用户想要筛选数据。
        当前 DataFrame 的列名是: {list(self.current_df.columns)}
        用户的筛选条件是: {condition}

        请生成一行 Python 代码来完成筛选，要求：
        1. 变量名必须是 'df'
        2. 返回筛选后的 DataFrame
        3. 只输出代码，不要有任何解释

        示例: df[df['price'] > 5000]
        """
        
        # 调用LLM生成代码
        response = self.llm.complete(prompt)
        generated_code = response.text.strip()
        
        # 清理代码（去掉可能的markdown标记）
        if generated_code.startswith("```"):
            generated_code = generated_code.split("\n")[1]
        if generated_code.endswith("```"):
            generated_code = generated_code.rsplit("\n", 1)[0]
        generated_code = generated_code.strip()
        
        print(f"生成的代码: {generated_code}")
        
        # 执行代码并更新 current_df
        try:
            df = self.current_df  # 为了代码能找到 df 变量
            filtered_df = eval(generated_code)   # 执行代码
            self.current_df = filtered_df    # 更新当前工作表
            print(f"✅ 筛选成功: {rows_before} 行 → {len(self.current_df)} 行")
        except Exception as e:
            print(f"❌ 代码执行失败: {e}")
            print(f"保持原数据不变")
        
        # 记录操作
        self.operation_history.append({
            'step': len(self.operation_history) + 1,
            'operation': 'FILTER',
            'description': condition,
            'code': generated_code,
            'rows': f"{rows_before} → {len(self.current_df)}"
        \})
        
        return self.current_df
    
    def aggregate(self, aggregation: str) -> str:
        """聚合操作 - 基于当前表计算"""
        print(f"[聚合] {aggregation}")
        
        # 构造prompt
        prompt = f"""
        你是一位 Pandas 专家。
        当前 DataFrame 有 {len(self.current_df)} 行数据，列名是: {list(self.current_df.columns)}
        用户想要: {aggregation}

        请生成一行 Python 代码来完成计算，要求：
        1. 变量名必须是 'df'
        2. 返回计算结果（数值或Series）
        3. 只输出代码，不要解释

        示例: df['rating'].mean()
        """
        
        response = self.llm.complete(prompt)   # 调用LLM生成代码
        generated_code = response.text.strip()  # 去除前后的空格
        
        # 清理代码
        if generated_code.startswith("```"):
            generated_code = generated_code.split("\n")[1]
        if generated_code.endswith("```"):
            generated_code = generated_code.rsplit("\n", 1)[0]
        generated_code = generated_code.strip()
        
        print(f"生成的代码: {generated_code}")
        
        # 执行计算
        try:
            df = self.current_df
            result = eval(generated_code)
            print(f"✅ 计算结果: {result}")
        except Exception as e:
            result = f"计算失败: {e}"
            print(f"❌ {result}")
        
        # 记录操作
        self.operation_history.append({
            'step': len(self.operation_history) + 1,
            'operation': 'AGGREGATE',
            'description': aggregation,
            'code': generated_code,
            'result': str(result)[:100]
        \})
        
        return str(result)
    
    def reset(self):
        """重置到原始表"""
        print("🔄 [重置] 恢复原始表格")
        self.current_df = self.original_df.copy()
        self.operation_history = []
    
    def show_history(self):
        """显示操作历史"""
        print("\n📜 操作历史:")
        print("-" * 60)
        for op in self.operation_history:
            print(f"  步骤 {op['step']}: {op['operation']}")
            print(f"    描述: {op['description']}")
            print(f"    代码: {op.get('code', 'N/A')}")
            if 'rows' in op:
                print(f"    行数变化: {op['rows']}")
            if 'result' in op:
                print(f"    结果: {op['result']}")
            print()

# 重新初始化
cot_engine = ChainOfTablesEngine(df)
print("✅ ChainOfTablesEngine 初始化完成（修正版）")

✅ ChainOfTablesEngine 初始化完成（修正版）

print("=" * 60)
print("🔗 Chain of Tables 实战示例")
print("=" * 60)

# 场景: 分析高端产品
print("\n📌 任务: 找出高价产品的平均评分")
print("思维链: 筛选(价格>5000) → 聚合(平均评分)")

# 步骤1: 筛选
print("\n" + "-" * 40)
cot_engine.filter("价格(price)大于5000")

# 步骤2: 聚合
print("\n" + "-" * 40)
result = cot_engine.aggregate("计算评分(rating)的平均值")
print(f"\n🎯 最终结果: {result}")

# 显示完整历史
cot_engine.show_history()

# 重置并执行另一个分析
print("\n" + "=" * 60)
print("📌 任务: 各类别产品数量排名")
print("思维链: 分组 → 计数 → 排序")
print("=" * 60)

cot_engine.reset()
result = cot_engine.aggregate("按类别(category)分组，计算每个类别的产品数量，按降序排列")
print(f"\n🎯 结果: {result}")
cot_engine.show_history()

============================================================ 🔗 Chain of Tables 实战示例

📌 任务: 找出高价产品的平均评分 思维链: 筛选(价格>5000) → 聚合(平均评分)

---

[筛选] 价格(price)大于5000 生成的代码: df[df['price'] > 5000] ✅ 筛选成功: 32 行 → 13 行

---

[聚合] 计算评分(rating)的平均值 生成的代码: df['rating'].mean() ✅ 计算结果: 4.176923076923076

🎯 最终结果: 4.176923076923076

📜 操作历史:

步骤 1: FILTER 描述: 价格(price)大于5000 代码: df[df['price'] > 5000] 行数变化: 32 → 13

步骤 2: AGGREGATE 描述: 计算评分(rating)的平均值 代码: df['rating'].mean() 结果: 4.176923076923076

============================================================ 📌 任务: 各类别产品数量排名 思维链: 分组 → 计数 → 排序

🔄 [重置] 恢复原始表格 [聚合] 按类别(category)分组，计算每个类别的产品数量，按降序排列 生成的代码: df.groupby('category').size().sort_values(ascending=False) ✅ 计算结果: category 手机 6 笔记本 6 .. 键盘 5 鼠标 5 Length: 6, dtype: int64

🎯 结果: category 手机 6 笔记本 6 .. 键盘 5 鼠标 5 Length: 6, dtype: int64

📜 操作历史:

步骤 1: AGGREGATE 描述: 按类别(category)分组，计算每个类别的产品数量，按降序排列 代码: df.groupby('category').size().sort_values(ascending=False) 结果: category 手机 6 笔记本 6 .. 键盘 5 鼠标 5 Length: 6, dtype: int64

# 数据验证
df1 = df[df['price'] > 5000]
df1

<div> <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; }

.dataframe tbody tr th { vertical-align: top; }

.dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>product_id</th> <th>name</th> <th>description</th> <th>tags</th> <th>price</th> <th>category</th> <th>stock</th> <th>brand</th> <th>rating</th> </tr> </thead> <tbody> <tr> <th>0</th> <td>1</td> <td>MacBook Pro 14英寸</td> <td>MacBook Pro 14英寸是一款旗舰级的笔记本产品，续航持久，游戏体验佳。</td> <td>轻薄, 办公, 游戏, 高性能</td> <td>15087</td> <td>笔记本</td> <td>25</td> <td>苹果</td> <td>3.6</td> </tr> <tr> <th>1</th> <td>2</td> <td>ThinkPad X1 Carbon</td> <td>ThinkPad X1 Carbon采用机械轴体技术，配置强大，使用便捷，适合设计师使用。</td> <td>轻薄, 设计, 办公</td> <td>7611</td> <td>笔记本</td> <td>398</td> <td>联想</td> <td>4.7</td> </tr> <tr> <th>...</th> <td>...</td> <td>...</td> <td>...</td> <td>...</td> <td>...</td> <td>...</td> <td>...</td> <td>...</td> <td>...</td> </tr> <tr> <th>12</th> <td>13</td> <td>iPad Pro</td> <td>iPad Pro是一款旗舰级的平板产品，屏幕出色，办公效率高。</td> <td>笔记, 娱乐</td> <td>6003</td> <td>平板</td> <td>118</td> <td>苹果</td> <td>4.9</td> </tr> <tr> <th>16</th> <td>17</td> <td>联想小新Pad Pro</td> <td>这款联想小新Pad Pro手感舒适，时尚美观且办公效率高，是平板中的佼佼者。</td> <td>绘画, 阅读, 轻办公</td> <td>7465</td> <td>平板</td> <td>107</td> <td>三星</td> <td>4.3</td> </tr> </tbody> </table>

13 rows × 9 columns

</div>

df1['rating'].mean()

np.float64(4.176923076923076)

# 将按类别(category)分组，计算每个类别的产品数量，按降序排列，第二个任务的结果获取出
query_result = df.groupby('category').size().sort_values(ascending=False)
query_result

category 手机 6 笔记本 6 .. 键盘 5 鼠标 5 Length: 6, dtype: int64

#可以通过matplotlib来对生成的Pandas数据框进行可视化
import matplotlib.pyplot as plt
import platform

# 1. 解决中文显示问题
system_name = platform.system()
if system_name == 'Darwin':  # macOS
    plt.rcParams['font.sans-serif'] = ['Arial Unicode MS']
elif system_name == 'Windows': # Windows
    plt.rcParams['font.sans-serif'] = ['SimHei']
else: # Linux
    plt.rcParams['font.sans-serif'] = ['WenQuanYi Micro Hei']
plt.rcParams['axes.unicode_minus'] = False 

# 2. 定义标题
title = '各类别产品数量分布'

# 3. 绘图
plt.figure(figsize=(10, 6))
ax = query_result.plot(kind='bar', color='steelblue')
plt.title(title)
plt.xlabel('类别')
plt.ylabel('数量')
plt.xticks(rotation=45, ha='right')

# 4. 添加数值标签
for p in ax.patches:
    ax.annotate(str(p.get_height()), (p.get_x() + p.get_width() / 2., p.get_height()),
                ha='center', va='center', xytext=(0, 5), textcoords='offset points')

plt.tight_layout()
plt.show()

3.4 PandasQueryEngine 总结

适用场景对比

数据规模	推荐方案	原因
< 1万行	PandasQueryEngine	零延迟、精准计算
1-10万行	Text-to-SQL	数据库优化、索引支持
> 10万行	Hybrid Retrieval	向量检索+过滤

优化技巧

1. 指令工程: 提供清晰的 instruction_str 指导 LLM
2. 错误处理: 实现重试机制应对 LLM 偶发错误
3. 性能监控: 记录查询耗时和成功率
4. 复杂问题: 使用 Chain of Tables 分解为简单步骤

局限性

• ⚠️ 大数据量时内存开销大
• ⚠️ 复杂查询可能生成错误代码
• ⚠️ 不支持跨表 JOIN（需预处理）