Prompt 工程:喂给 LLM 的到底是什么
30 秒导读: 你问 PandasAI 一个关于数据的自然语言问题,它并不会把整张表倒给大模型。 它用 Jinja2 模板拼一段 prompt——里面有表的结构和几行样本、一个已经帮你写好的
execute_sql_query()函数、一段要你填空的初始 Python 骨架、一条输出必须长成什么样 的硬约束,外加从向量库召回的相似问答。LLM 顺着这段 prompt 吐回一段 Python,PandasAI 再从回复里把代码抠出来。这一章讲的就是这段输入到底怎么组装、每一块为什么在那里。
本章只讲「LLM 实际看到的输入怎么来的」。代码抠出来之后怎么校验、清洗、安全执行,是 代码即 SQL 的事;数据集/视图/SQL 编译在 语义层; 整个一问一答的主循环在 Agent 主循环。本章尽量不重复它们。
1. 这是什么(零基础也能懂)
一句话: prompt 工程 = 把「用户的问题」翻译成「一段能让 LLM 稳定写出正确代码的文字」。
同一个问题,喂给模型的文字组织得好不好,直接决定它写的代码能不能跑、结果对不对。PandasAI 的核心设计是:别让模型硬啃原始数据,而是给它足够的"上下文 + 规矩",让它写代码去查数据。
它给 LLM 喂的东西,可以拆成五类:
| 喂进去的东西 | 白话 | 为什么要 |
|---|---|---|
| 表结构 + 几行样本 | "你要查的表叫什么、有哪些列、长什么样" | 模型没见过你的数据,得先"看一眼" |
execute_sql_query() 函数 | "查数据的工具我给你写好了,直接调" | 统一入口,模型不用自己连库 |
| 初始/复用代码骨架 | "照着这个模板改" 或 "在你上次的代码上接着改" | 约束输出结构,支持多轮对话 |
| 输出格式约束 | "最后必须给我一个 {type, value} 的 result" | 结果能被程序解析 |
| RAG 召回的问答/文档 | "这是几个相 似的例子,参考着写" | 用你训练过的样例纠偏 |
一句话直觉: 把它想成给一个很聪明但从没见过你数据库的实习生派活——你不会把百万行 数据打印给他,你会给他:表的字段说明、几行样本、一个封装好的查询函数、一个代码模板,和 一句"记住:聚合排序 join 都用 SQL 做"。PandasAI 干的就是自动写这张"任务交接单"。
2. 顶层全景(一段 prompt 怎么被拼出来)
一次提问,从"用户字符串"到"LLM 看到的完整 prompt"再到"抠出的 Python",走的是这条线:
用户问题 str
│ agent.base.generate_code() 把问题 add 进 memory(is_user=True)
▼
get_chat_prompt_for_sql(state) prompts/__init__.py:19
│ 造出 GeneratePythonCodeWithSQLPrompt(带 context / last_code / output_type)
▼
BasePrompt.__init__ 用 Jinja2 载入 .tmpl prompts/base.py:23
│
▼
prompt.to_string() → Jinja 渲染主模板 generate_python_code_with_sql.tmpl
├─ <tables> 块 ── 每个 df 调 serialize_dataframe() ──► <table …>+head() CSV
├─ sql_functions ── 给出 execute_sql_query 签名
├─ 初始代码骨架 vs 复用 last_code_generated(看是否多轮)
├─ output_type_template ── 输出必须是 {type,value}
├─ vectordb_docs ── 从向量库召回相似 Q&A / docs
└─ 末尾硬约束 ── "聚合/排序/join/groupby 都走 SQL"
▼
llm.generate_code(prompt, ctx) llm/base.py:161
│ call() 打模型 → _extract_code() / _polish_code() 抠出 Python
▼
一段干净的 Python code(交给第 2 章去校验/清洗/执行)
各部件一句话职责:
| 部件 | 干什么 | 在哪 |
|---|---|---|
BasePrompt | prompt 基类:用 Jinja2 渲染,to_json 拼对话+system | core/prompts/base.py:14 |
get_chat_prompt_for_sql | 主 prompt 工厂;还有两个纠错 prompt 工厂 | core/prompts/__init__.py:19 |
*.tmpl 主模板 | 决定 prompt 长啥样(tables/函数/骨架/约束) | core/prompts/templates/generate_python_code_with_sql.tmpl |
DataframeSerializer | 把 DataFrame 压成 <table> + 截断的 CSV | helpers/dataframe_serializer.py:8 |
LLM.generate_code | 打模型 + 从回复里抠 Python | llm/base.py:161 |
GenerateSystemMessagePrompt | 组 system prompt(身份 + 历史对话) | core/prompts/generate_system_message.py:4 |
VectorStore | train() 存的 Q&A/文档,渲染时召回 | vectorstores/vectorstore.py:5 |
3. 核心原理(逐块拆)
3.1 BasePrompt:一个 Jinja2 模板的两种活法
它要解决的小问题: prompt 里有大量"看情况填/不填"的分支(多轮就复用旧代码、有向量库 就召回、有 output_type 就换约束)。手写字符串拼接会乱成一团,所以用模板引擎。
思路: 每个 prompt 是一个 BasePrompt 子类。它要么带内联 template 字符串、要么带
template_path 指向一个 .tmpl 文件;构造时用 Jinja2 载入,渲染时把 **kwargs(叫 props)
灌进去。
真实实现,core/prompts/base.py:23 BasePrompt.__init__——两条载入路径:
if self.template: # 内联字符串模板
env = Environment()
self.prompt = env.from_string(self.template)
elif self.template_path: # 从 templates/ 目录载 .tmpl 文件
path_to_template = os.path.join(Path(__file__).parent, "templates")
env = Environment(loader=FileSystemLoader(path_to_template))
self.prompt = env.get_template(self.template_path)
渲染入口 to_string()(base.py:48)只是 self.prompt.render(**self.props),并缓存到
_resolved_prompt。注意 render()(base.py:39)另有一个会把 3 个以上连续换行压成 2 个
的清洗(re.sub(r"\n{3,}", "\n\n", render))——因为模板里大量 {% if %} 分支不命中会留下
空行,不压一下 prompt 会很难看。
关键细节: 主 prompt 用的是 template_path(见 3.2),纠错 prompt 也是。真正的差异全在
.tmpl 文件里,Python 类几乎是空壳(GeneratePythonCodeWithSQLPrompt 就一行 template_path,
generate_python_code_with_sql.py:4)。
3.2 三个 prompt 工厂:一个主 prompt + 两个"救火" prompt
它要解决的小问题: 正常提问要一个 prompt;代码执行报错了、或者输出类型不对,还得能 带着错误信息重新问一遍。PandasAI 把这三种情况各封装成一个工厂函数。
core/prompts/__init__.py 里的三个工厂:
| 工厂函数 | 何时用 | 造出的 prompt | 关键入参 |
|---|---|---|---|
get_chat_prompt_for_sql (:19) | 每次正常提问 | GeneratePythonCodeWithSQLPrompt | last_code_generated、output_type |
get_correct_error_prompt_for_sql (:27) | 代码跑出异常 | CorrectExecuteSQLQueryUsageErrorPrompt | code、error(traceback) |
get_correct_output_type_error_prompt (:35) | 输出类型不符 | CorrectOutputTypeErrorPrompt | code、error、output_type |
主 prompt 的调用点在 agent/base.py:117:先把用户问题 memory.add(..., is_user=True),再
get_chat_prompt_for_sql(self._state) 造 prompt,交给 CodeGenerator.generate_code。
两个纠错 prompt 的模板很像,都是"把表结构 + 函数签名 + 原问题 + 你写的代码 + 报错"拼一遍,
末尾要求重写。区别只在最后一句:执行错版本要求"新代码仍要用 execute_sql_query 函数"
(correct_execute_sql_query_usage_error_prompt.tmpl:14),输出类型错版本要求"结果类型必须是
{{output_type}}"(correct_output_type_error_prompt.tmpl:14)。这套重试怎么被驱动,见
第 2 章 与 第 1 章。
3.3 主模板:LLM 看到的 prompt 长这样
它要解决的小问题: 把上面那五类东西按固定顺序、按当前上下文,拼成一段完整文字。
主模板 templates/generate_python_code_with_sql.tmpl 全文很短,是理解本章的核心。逐块看:
(a) tables 块 —— 遍历所有 df,每个都渲染一次 shared/dataframe.tmpl,而后者只有一句
{{ df.serialize_dataframe() }}(见 3.4):
<tables>
{% for df in context.dfs %}
{% include 'shared/dataframe.tmpl' with context %}
{% endfor %}
</tables>
(b) sql_functions 块 (shared/sql_functions.tmpl:1) —— 告诉模型"查询工具已备好,别重定义":
The following functions have already been provided. Please use them as needed and do not redefine them.
<function>
def execute_sql_query(sql_query: str) -> pd.DataFrame
"""This method connects to the database, executes the sql query and returns the dataframe"""
</function>
若 context.skills 非空,还会把每个 skill 追加进来。这一步是整个"代码即 SQL"设计的支点:
模型不会自己 pd.read_csv,它只会写 SQL 字符串然后 execute_sql_query(...)。
(c) 初始骨架 vs 复用旧代码 —— 这是多轮对话的关键分支(.tmpl:9-22):
{% if last_code_generated and context.memory.count() > 0 %}
Last code generated:
{{ last_code_generated }}
{% else %}
Update this initial code:
```python
# TODO: import the required dependencies
import pandas as pd
# Write code here
# Declare result var: {% include 'shared/output_type_template.tmpl' with context %}
```
{% endif %}
首轮(memory 里没历史)给一段"填空模板";后续轮把上一轮生成的代码塞进来,让模型在旧代码
上增量修改——这就是 PandasAI 多轮对话"记得上文"的机制之一。last_code_generated 由
CodeGenerator.generate_code 在每次生成后写回 context(core/code_generation/base.py:35,41)。
(d) vectordb_docs 块 —— RAG 召回,见 3.6。
(e) 最后一条消息 + 输出约束 + 硬约束 —— .tmpl:24-33:
{{ context.memory.get_last_message() }}
At the end, declare "result" variable as a dictionary of type and value in the following format:
{% include 'shared/output_type_template.tmpl' with context %}
Generate python code and return full updated code:
### Note: Use only relevant table for query and do aggregation, sorting, joins and grouby through sql query
末尾那句 ### Note: 是整份 prompt 的硬约束:聚合、排序、join、groupby 一律走 SQL。这句话
把"重活下推到数据库"从设计意图变成了对模型的明确指令——模型被引导只用 Python 做壳、用 SQL
做真正的数据处理。
3.4 DataframeSerializer:一张表怎么被压进 prompt
它要解决的小问题: 表可能有百万行、某些单元格是长文本或 JSON。全塞进 prompt 会爆 token, 所以要只给结构 + 少量样本 + 截断长文本。
DataframeSerializer.serialize(helpers/dataframe_serializer.py:11)产出一段
<table …> 标签 + head() 的 CSV。它拼出的东西大概长这样:
<table dialect="postgres" table_name="sales" description="…"
columns="[{...}]" dimensions="1000000x5">
id,amount,note
1,120,short text
2,88,another…
</table>
拆开看它塞了什么:
| 属性 | 来源 | 作用 |
|---|---|---|
dialect | df.get_dialect()(见下) | 告诉模型该写哪种方言的 SQL |
table_name | df.schema.name | SQL 里 FROM 用哪个表名 |
description | df.schema.description(可选) | 给模型语义提示 |
columns | df.schema.columns 逐个 model_dump() 后 JSON | 列名/类型等结构 |
dimensions | df.rows_countxdf.columns_count | 让模型知道数据规模 |
| CSV 正文 | df.head() 再截断 | 只给前几行样本,不给全量 |
两个易忽略但重要的细节:
- 只序列化
head()。 第 38 行cls._truncate_dataframe(df.head())——喂进 prompt 的永远 只是表头几行,不是全表。模型靠这几行 + 列结构去"想象"数据,真正的数据处理留给 SQL。 - 长文本截断。
_truncate_dataframe(:48)对每个单元格:dict/list 先json.dumps成 字符串,超过MAX_COLUMN_TEXT_LENGTH = 200(:9)就切到 200 字符再加个省略号…(:56-57)。防止一个超长字段撑爆整段 prompt。
dialect 从哪来?dataframe/base.py:135 get_dialect():有 source 就用 source.type(本地源
如 csv 走 duckdb),没有 source 默认 postgres——与 serializer 的默认值一致。
3.5 从模型回复里"抠"出 Python:_extract_code / _polish_code
它要解决的小问题: 模型的回复往往不是纯代码——外面裹着 ```python、解释文字、
反引号。得可靠地把中间那段 Python 拿出来。
链路是 LLM.generate_code(llm/base.py:161)→ 抽象方法 call() 打模型 → _extract_code()。
_extract_code(llm/base.py:94)的逻辑:
- 如果回复里有分隔符
```(默认),就取第一对分隔符之间的内容:code.split(separator)[1](:113)。 - 交给
_polish_code清理。 - 最后用
ast.parse判断是不是合法 Python(_is_python_code,:79);不是就抛NoCodeFoundError(:117-118)——宁可报错,也不把非代码硬塞给执行器。
_polish_code(llm/base.py:60)做三件小事:去掉开头的 python/py 前缀、剥掉整体被单个
反引号包住的情况、strip() 首尾空白。都是针对模型输出的常见"脏格式"。
关键细节: 这里只负责"从文本里拿到一段合法 Python 字符串"。拿到之后的危险 API 校验、 清洗、沙箱执行是另一回事,全在 第 2 章。
3.6 system prompt 与记忆注入
它要解决的小问题: 有些模型分 system / user 消息,有些不分;还要把 agent 的身份描述和 历史对话带进去。
记忆(Memory) 是一个存 {message, is_user} 列表的小类(helpers/memory.py:5)。它按
memory_size 只保留最近 N 条,并有一堆取数方法:get_last_message()(最后一条,主模板用)、
get_previous_conversation()(除最后一条外的历史)、to_json()(转成 role/message 列表)。
注意 get_messages(:39)会给助手的历史回 复做 100 字符截断(_truncate),防历史撑大 prompt。
system prompt 由 GenerateSystemMessagePrompt(generate_system_message.py:4)渲染
generate_system_message.tmpl:
{% if memory.agent_description %} {{memory.agent_description}} {% endif %}
{% if memory.count() > 1 %}
### PREVIOUS CONVERSATION
{{ memory.get_previous_conversation() }}
{% endif %}
即 "agent 身份描述" + "以往对话"。对不支持 messages 的模型,LLM.prepend_system_prompt
(llm/base.py:122)干脆把 system prompt 字符串拼在 user prompt 前面;对支持的,则走
Memory.to_openai_messages()(memory.py:81)组成带 system role 的消息数组。
两种打包方式 由 BasePrompt.to_json(base.py:61)体现:没有 context 就只返回
{"prompt": ...};有 context 则返回 {"conversation", "system_prompt", "prompt"} 三件套
——conversation 来自 memory.to_json(),system_prompt 来自 memory.agent_description。
主 prompt 子类还覆写了 to_json,额外带上 datasets 和 config(direct_sql / output_type,
见 generate_python_code_with_sql.py:9)。
3.7 训练 / RAG:train() 与向量库召回
它要解决的小问题: 通用模型不了解你的业务语义。给它几个"标准问答对"或参考文档当例子,
能显著提升生成代码的准确度。这就是 PandasAI 的 train()。
写入端——Agent.train(agent/base.py:218):
- 必须先配了 vectorstore,否则抛
MissingVectorStoreError(:233-236)。 queries和codes要么都给、要么都不给(:238-241),给了就vectorstore.add_question_answer(queries, codes)(:247)。- 额外文档走
vectorstore.add_docs(docs)(:244)。
VectorStore 是抽象接口(vectorstores/vectorstore.py:5),定义了 add_question_answer、
add_docs、get_relevant_qa_documents、get_relevant_docs_documents 等;问答对渲染成文本时
用 _format_qa(:178)拼成 "Q: {query}\n A: {code}"。
召回端——渲染主 prompt 时 shared/vectordb_docs.tmpl 用当前问题去向量库找相似内容:
{% if context.vectorstore %}
{% set documents = context.vectorstore.get_relevant_qa_documents(context.memory.get_last_message()) %}
{% if documents|length > 0 %}You can utilize these examples as a reference for generating code.{% endif %}
{% for document in documents %}{{ document }}{% endfor %}
{% endif %}
… 同理再召回 get_relevant_docs_documents,前缀 "Here are additional documents for reference." …
两轮召回:先找相似问答对(带一句"可参考这些例子"),再找相似文档(带一句"这些额外
文档可参考")。检索的 query 都是 memory.get_last_message()——也就是用户刚问的那句。没配
向量库时整块 {% if %} 不命中、什么都不加(空行随后被 render() 压掉)。
4. 巧妙之处(可借鉴)
-
把"给工具"变成"给函数签名"。 prompt 里直接写好
def execute_sql_query(...)的签名 + docstring + "不要重定义",比抽象地说"你可以查询数据库"精确得多——模型照着签名调就行 (shared/sql_functions.tmpl:2-5)。 -
只喂
head()+ 截断长文本。 用极小的 token 成本让模型"看见"数据形状,把真正的数据量 留给 SQL 处理(dataframe_serializer.py:38,56)。这是"代码即 SQL"能省 token 的直接原因。 -
末尾一句硬约束定调。 一行
### Note: … do aggregation, sorting, joins and grouby through sql query(generate_python_code_with_sql.tmpl:33)就把整个架构意图压给了模型。 -
多轮 = 把上一轮代码塞回 prompt。 不靠复杂状态机,靠
last_code_generated分支 (.tmpl:9)让模型在旧代码上增量改,简单且有效。 -
抠代码宁缺毋滥。
_extract_code拿到东西后还要ast.parse验一遍,不是合法 Python 就抛错(llm/base.py:117),而不是把垃圾传给下游执行。
5. 边界与局限
-
只序列化
head()意味着模型对"分布/异常值/全量特征"是盲的;它只能靠列结构 + 表头几行 推断,复杂语义仍依赖description、train()的例子来补(dataframe_serializer.py:38)。 -
_extract_code只取第一对分隔符之间(split(separator)[1],llm/base.py:113)。如果 模型回复里在正 式代码前先放了一个无关的代码块,抠出来的可能是错的那段。 -
长文本按 200 字符硬截断(
MAX_COLUMN_TEXT_LENGTH,:9)。对需要看完整文本才能判断的 任务(如长描述里藏着关键词),截断可能丢信息。 -
RAG 召回 query 固定是最后一条用户消息(
vectordb_docs.tmpl),多轮里指代不清的追问 ("那再按地区分组呢")召回效果会打折。
6. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| Jinja2 载入模板 / 渲染 | pandasai/core/prompts/base.py | BasePrompt.__init__、render、to_string |
| 对话+system 打包 | pandasai/core/prompts/base.py | BasePrompt.to_json |
| 三个 prompt 工厂 | pandasai/core/prompts/__init__.py | get_chat_prompt_for_sql、get_correct_error_prompt_for_sql、get_correct_output_type_error_prompt |
| 主 prompt 类 | pandasai/core/prompts/generate_python_code_with_sql.py | GeneratePythonCodeWithSQLPrompt |
| 主模板(tables/函数/骨架/约束) | pandasai/core/prompts/templates/generate_python_code_with_sql.tmpl | — |
| execute_sql_query 签名块 | pandasai/core/prompts/templates/shared/sql_functions.tmpl | — |
| 输出格式约束 | pandasai/core/prompts/templates/shared/output_type_template.tmpl | — |
| RAG 召回块 | pandasai/core/prompts/templates/shared/vectordb_docs.tmpl | — |
| DataFrame 序列化 | pandasai/helpers/dataframe_serializer.py | DataframeSerializer.serialize、_truncate_dataframe |
| 方言选择 / 触发序列化 | pandasai/dataframe/base.py | get_dialect、serialize_dataframe |
| 抠代码 / 清理代码 | pandasai/llm/base.py | generate_code、_extract_code、_polish_code、_is_python_code |
| system prompt 拼装 | pandasai/llm/base.py | get_system_prompt、prepend_system_prompt |
| system message 模板类 | pandasai/core/prompts/generate_system_message.py | GenerateSystemMessagePrompt |
| 记忆存取 | pandasai/helpers/memory.py | Memory.get_last_message、get_previous_conversation、to_json、to_openai_messages |
| 主 prompt 调用点 | pandasai/agent/base.py | Agent.generate_code |
| 训练写入向量库 | pandasai/agent/base.py | Agent.train |
| 向量库接口 | pandasai/vectorstores/vectorstore.py | VectorStore.add_question_answer、get_relevant_qa_documents、_format_qa |