表格数据的 AI 理解——结构化数据怎么喂给大模型
2026/4/30大约 9 分钟
表格数据的 AI 理解——结构化数据怎么喂给大模型
适读人群：做数据分析 AI、企业报表问答的开发者 | 阅读时长：约 16 分钟 | 核心价值：掌握表格数据送给 LLM 的三种工程方案，知道各自的适用场景
我有个做运营的朋友，她们公司上了一套 AI 问答系统。
有一天她跟我说：这个 AI 太蠢了，我问它"上周销售额最高的品类是什么"，它说"根据数据，销售额表现良好"，完全没有给出答案。
她让我帮她看看。我去看了他们的实现，发现了问题所在：
工程师把 Excel 表格直接读成文本，然后往模型里一丢：
品类 销售额 环比 同比 女装 125万 +8% +15% 男装 98万 -3% +7%
鞋靴 87万 +12% +22% 箱包 45万 +5% +8% ...所有数据变成了一行字符串，表格结构完全丢失。模型当然不知道哪个数字对应哪个品类，更别说做比较分析了。
这就是"把数据喂给 LLM"这件事里最核心的坑：你以为你给了它数据，其实给了它一堆乱码。
三种主要方案
处理表格数据有三个核心方向，适用场景不同，不存在哪个方案"最好"。
方案一：Text2SQL——让 LLM 生成 SQL 查询
适用场景：数据在关系型数据库里，表结构固定，需要支持灵活的查询。
Text2SQL 的核心思路是：告诉模型你有哪些表、字段是什么意思，然后让它把用户问题翻译成 SQL，执行 SQL 拿到结果，再让模型解读结果。
import anthropic
import sqlite3
import json
from typing import Optional
class Text2SQLAgent:
"""
Text2SQL 问答 Agent
"""
def __init__(self, db_path: str, schema_description: dict):
"""
db_path: 数据库路径
schema_description: 表结构描述，格式见下面的示例
"""
self.db_path = db_path
self.schema = schema_description
self.client = anthropic.Anthropic()
def _build_schema_prompt(self) -> str:
"""把表结构描述转换为模型能理解的文本"""
lines = ["数据库表结构：\n"]
for table_name, table_info in self.schema.items():
lines.append(f"表名：{table_name}")
lines.append(f"说明：{table_info.get('description', '')}")
lines.append("字段：")
for col in table_info['columns']:
col_desc = f" - {col['name']} ({col['type']}): {col['description']}"
if col.get('example'):
col_desc += f"，例：{col['example']}"
lines.append(col_desc)
lines.append("")
return '\n'.join(lines)
def generate_sql(self, user_question: str) -> dict:
"""根据用户问题生成 SQL"""
schema_text = self._build_schema_prompt()
response = self.client.messages.create(
model="claude-opus-4-5",
max_tokens=1000,
messages=[{
"role": "user",
"content": f"""{schema_text}
用户问题：{user_question}
请生成 SQLite 格式的 SQL 查询语句来回答这个问题。
要求：
1. 只返回 SQL，不要解释
2. SQL 要能直接执行
3. 如果问题涉及"最高"、"最低"、"排名"等，用 ORDER BY + LIMIT
4. 日期相关用 SQLite 的日期函数
5. 如果问题无法用现有表结构回答，返回：CANNOT_ANSWER
SQL:"""
}]
)
sql = response.content[0].text.strip()
# 安全检查：只允许 SELECT
if not sql.upper().startswith('SELECT') and sql != 'CANNOT_ANSWER':
return {'success': False, 'error': '只允许查询操作'}
return {'success': True, 'sql': sql}
def execute_sql(self, sql: str) -> dict:
"""执行 SQL，返回结果"""
try:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row # 让结果可以用字段名访问
cursor = conn.cursor()
cursor.execute(sql)
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
data = [dict(zip(columns, row)) for row in rows]
conn.close()
return {'success': True, 'data': data, 'columns': columns, 'row_count': len(data)}
except Exception as e:
return {'success': False, 'error': str(e)}
def interpret_results(self, question: str, sql: str, query_result: dict) -> str:
"""让模型解读查询结果，生成自然语言回答"""
if not query_result['data']:
data_text = "查询结果为空"
else:
# 把结果格式化为表格文字
data_text = json.dumps(query_result['data'], ensure_ascii=False, indent=2)
if len(data_text) > 3000:
# 数据太多，只显示前几行
data_text = json.dumps(query_result['data'][:20], ensure_ascii=False, indent=2)
data_text += f"\n...（共 {query_result['row_count']} 条结果，只显示前20条）"
response = self.client.messages.create(
model="claude-opus-4-5",
max_tokens=500,
messages=[{
"role": "user",
"content": f"""用户问题：{question}
执行的 SQL：{sql}
查询结果：
{data_text}
请用简洁的中文回答用户的问题。直接给出答案，不需要解释 SQL 是怎么写的。"""
}]
)
return response.content[0].text
def answer(self, question: str) -> str:
"""完整的问答流程"""
# 生成 SQL
sql_result = self.generate_sql(question)
if not sql_result['success']:
return f"无法处理这个问题：{sql_result['error']}"
sql = sql_result['sql']
if sql == 'CANNOT_ANSWER':
return "抱歉，这个问题超出了我能查询的数据范围。"
# 执行查询
query_result = self.execute_sql(sql)
if not query_result['success']:
# SQL 有问题，让模型修复
return self._retry_with_error(question, sql, query_result['error'])
# 解读结果
return self.interpret_results(question, sql, query_result)
def _retry_with_error(self, question: str, failed_sql: str, error: str) -> str:
"""SQL 执行失败时，让模型修复"""
response = self.client.messages.create(
model="claude-opus-4-5",
max_tokens=500,
messages=[{
"role": "user",
"content": f"""以下 SQL 执行出错：
SQL: {failed_sql}
错误: {error}
原始问题: {question}
{self._build_schema_prompt()}
请修复 SQL（只返回修复后的 SQL）:"""
}]
)
fixed_sql = response.content[0].text.strip()
query_result = self.execute_sql(fixed_sql)
if not query_result['success']:
return "抱歉，无法处理这个查询，请换一种方式提问。"
return self.interpret_results(question, fixed_sql, query_result)
# 使用示例
schema = {
'sales': {
'description': '销售数据表，每行是一个品类在某天的销售数据',
'columns': [
{'name': 'date', 'type': 'DATE', 'description': '日期', 'example': '2024-01-15'},
{'name': 'category', 'type': 'TEXT', 'description': '品类名称', 'example': '女装'},
{'name': 'revenue', 'type': 'REAL', 'description': '销售额（元）'},
{'name': 'orders', 'type': 'INTEGER', 'description': '订单量'},
{'name': 'channel', 'type': 'TEXT', 'description': '销售渠道', 'example': '线上/线下'},
]
}
}
agent = Text2SQLAgent(db_path='sales.db', schema_description=schema)
print(agent.answer("上周销售额最高的品类是什么？"))
# 输出：上周销售额最高的品类是女装，销售额为 125.3 万元。方案二：表格序列化——把表格转成 LLM 友好的格式
适用场景：表格数据量不大（几十行以内），需要 LLM 做综合理解，不只是查询。
关键是序列化格式的选择，直接影响模型的理解效果：
import pandas as pd
from typing import Literal
def serialize_dataframe(
df: pd.DataFrame,
format: Literal['markdown', 'json', 'csv', 'description'] = 'markdown',
max_rows: int = 50,
context: str = ''
) -> str:
"""
把 DataFrame 序列化为 LLM 友好的格式
format 选择建议：
- markdown：适合大多数场景，可读性好
- json：适合需要精确解析的场景
- description：适合数据量大时的摘要
"""
# 如果数据太多，只取前 N 行
if len(df) > max_rows:
note = f"\n（注：原始数据共 {len(df)} 行，此处显示前 {max_rows} 行）"
display_df = df.head(max_rows)
else:
note = ''
display_df = df
if format == 'markdown':
return _to_markdown(display_df, context, note)
elif format == 'json':
return _to_json(display_df, context, note)
elif format == 'description':
return _to_description(df, context)
else:
return display_df.to_csv(index=False) + note
def _to_markdown(df: pd.DataFrame, context: str, note: str) -> str:
"""转换为 Markdown 表格格式"""
lines = []
if context:
lines.append(f"表格说明：{context}\n")
# 表头
headers = list(df.columns)
lines.append('| ' + ' | '.join(str(h) for h in headers) + ' |')
lines.append('| ' + ' | '.join(['---'] * len(headers)) + ' |')
# 数据行
for _, row in df.iterrows():
values = []
for val in row:
if pd.isna(val):
values.append('')
elif isinstance(val, float):
values.append(f'{val:.2f}')
else:
values.append(str(val))
lines.append('| ' + ' | '.join(values) + ' |')
return '\n'.join(lines) + note
def _to_description(df: pd.DataFrame, context: str) -> str:
"""
当数据量大时，转换为统计描述
让 LLM 理解数据的分布和规律，而不是具体每行
"""
lines = []
if context:
lines.append(f"数据说明：{context}\n")
lines.append(f"数据集概览：共 {len(df)} 行，{len(df.columns)} 列\n")
# 数值列的统计
numeric_cols = df.select_dtypes(include=['number']).columns
if len(numeric_cols) > 0:
lines.append("数值字段统计：")
stats = df[numeric_cols].describe()
lines.append(_to_markdown(stats.round(2), '', ''))
# 分类列的分布
categorical_cols = df.select_dtypes(include=['object']).columns
for col in categorical_cols[:5]: # 最多显示5个分类列
value_counts = df[col].value_counts().head(10)
lines.append(f"\n{col} 的值分布（Top 10）：")
for val, count in value_counts.items():
pct = count / len(df) * 100
lines.append(f" - {val}: {count} 次 ({pct:.1f}%)")
return '\n'.join(lines)
def answer_table_question(df: pd.DataFrame, question: str, table_context: str = '') -> str:
"""
用 LLM 回答关于表格的问题
"""
client = anthropic.Anthropic()
# 根据数据量选择序列化方式
if len(df) <= 30:
table_text = serialize_dataframe(df, format='markdown', context=table_context)
elif len(df) <= 200:
table_text = serialize_dataframe(df, format='markdown', max_rows=50, context=table_context)
else:
# 数据量大，先用统计描述，可能需要 Text2SQL
table_text = serialize_dataframe(df, format='description', context=table_context)
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=800,
messages=[{
"role": "user",
"content": f"""以下是数据表格：
{table_text}
请回答：{question}
直接回答问题，给出具体数字和结论。"""
}]
)
return response.content[0].text方案三：分析型问答——LLM 直接写分析代码执行
适用场景：需要复杂计算、统计分析，用自然语言指令驱动数据分析。
import pandas as pd
import io
import traceback
def code_interpreter_analysis(df: pd.DataFrame, question: str) -> dict:
"""
让 LLM 生成 Python 分析代码，在沙箱里执行
"""
client = anthropic.Anthropic()
# 提供数据的基本信息
df_info = {
'shape': df.shape,
'columns': list(df.columns),
'dtypes': df.dtypes.to_dict(),
'sample': df.head(3).to_dict(orient='records')
}
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=1500,
messages=[{
"role": "user",
"content": f"""你有一个 pandas DataFrame，变量名为 df。
数据信息：
- 行数: {df_info['shape'][0]}，列数: {df_info['shape'][1]}
- 列名: {df_info['columns']}
- 数据类型: {df_info['dtypes']}
- 前3行样本: {json.dumps(df_info['sample'], ensure_ascii=False)}
问题：{question}
请生成 Python 代码来回答这个问题。
要求：
1. 只使用 pandas 和内置模块，不要导入其他库
2. 把最终答案赋值给变量 `result`
3. `result` 可以是字符串、数字、列表或 DataFrame
4. 只返回 Python 代码，不要解释
\`\`\`python
# 你的代码
\`\`\`"""
}]
)
# 提取代码
code_text = response.content[0].text
code_match = re.search(r'```python\n(.*?)\n```', code_text, re.DOTALL)
if not code_match:
return {'success': False, 'error': '模型没有生成合法的 Python 代码'}
code = code_match.group(1)
# 在受限环境里执行代码
result = execute_analysis_code(df, code)
return result
def execute_analysis_code(df: pd.DataFrame, code: str) -> dict:
"""
在受限环境里执行分析代码
注意：这里只做演示，生产环境需要用更严格的沙箱
"""
local_vars = {'df': df.copy(), 'pd': pd, 'json': json}
try:
exec(code, {'__builtins__': {}}, local_vars)
result = local_vars.get('result', '代码执行成功但没有 result 变量')
# 如果结果是 DataFrame，转换为字典
if isinstance(result, pd.DataFrame):
result = result.to_dict(orient='records')
elif isinstance(result, pd.Series):
result = result.to_dict()
return {'success': True, 'result': result, 'code': code}
except Exception as e:
return {'success': False, 'error': str(e), 'code': code}三个方案怎么选
你的数据在哪里？
|
├── 关系型数据库
│ └── Text2SQL（最灵活，支持任意查询）
│
├── Excel / CSV 文件
│ ├── 数据量小（< 100行）→ 直接序列化 + LLM 问答
│ ├── 数据量中等（100-10000行）→ Text2SQL（先导入 SQLite）
│ └── 需要复杂统计计算 → Code Interpreter 方案
│
└── 实时报表
├── 查询需求简单 → 序列化 + LLM
└── 查询需求复杂 → Text2SQL一个容易犯的错误：数据量稍微大一点就想用序列化方案，直接把几千行数据塞给模型，既贵效果又差。100 行以上的表格，认真考虑 Text2SQL。
几个工程细节
字段名要有意义
col1、f_amt、dt这种字段名，模型很难理解。给模型提供字段的中文说明，Text2SQL 的准确率会提升很多。
数值格式要标注单位
"金额单位：元"、"时间格式：YYYY-MM-DD"这类说明要写清楚，不然模型可能生成按万元计算的 SQL，结果差了 10000 倍。
对 SQL 做安全检查
模型生成的 SQL 执行前，至少检查：是不是 SELECT、有没有危险函数（DROP、DELETE、UPDATE）、有没有访问不该访问的表。
结果为空要处理
查询返回空结果时，不要直接把空数组给模型，要明确告诉它"没有符合条件的数据"，不然模型可能会编造答案。
朋友那个系统后来怎么样了？
我帮他们改成了 Text2SQL 方案，加上合理的 schema 描述。用了一个星期，她反馈说"好多了，90% 的问题都能得到准确答案"。
剩下 10% 是什么？那是她们用自然语言问了一些模糊的、连她自己都不确定怎么定义的问题——这不是技术问题，是业务问题。
表格数据喂给 LLM，选对方案就成功了一半。