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_count`x`df.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),多轮里指代不清的追问 ("那再按地区分组呢")召回效果会打折。
代码抠出来之后的安全性/正确性不归本章——校验与清洗见第 2 章; 跑不通时的重试驱动见第 1 章。

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`

1. 这是什么(零基础也能懂)​

2. 顶层全景(一段 prompt 怎么被拼出来)​

3. 核心原理(逐块拆)​

3.1 BasePrompt:一个 Jinja2 模板的两种活法​

3.2 三个 prompt 工厂:一个主 prompt + 两个"救火" prompt​

3.3 主模板:LLM 看到的 prompt 长这样​

3.4 DataframeSerializer:一张表怎么被压进 prompt​

3.5 从模型回复里"抠"出 Python:_extract_code / _polish_code​

3.6 system prompt 与记忆注入​

3.7 训练 / RAG:train() 与向量库召回​

4. 巧妙之处(可借鉴)​

5. 边界与局限​

6. 代码地图(导航索引)​