关于 pixeltable
pixeltable 是面向构建数据密集型 AI 应用的 Python 框架,将表、计算列、媒体处理、向量索引和服务集中在应用文件中。该技能提供项目构建与命令使用指导,并强调以表格计算流程代替外部向量数据库和逐行模型循环。
主要功能
- 01以 TableModel 定义表结构
- 02插入时运行计算列链
- 03支持媒体拆分与处理
- 04通过 EmbeddingIndex 建立索引
- 05提供 HTTP 服务与命令行流程
适用场景
- 01构建基于文档的检索应用
- 02在表格中处理媒体与模型输出
- 03通过 FastAPI 路由提供应用服务
MCP 相关
文档提及 Developer MCP 和可选 Cloud MCP;后者可读取和操作 Cloud。
安装
复制下面的命令,在终端中运行。
npx skills add https:/ / github.com/ pixeltable/ pixeltable-skill --skill pixeltable安装后在 Claude Code 中输入/pixeltable确认已加载。
包含的文件
除 SKILL.md 外,安装时一起下载的参考资料和脚本。
agents/openai.yamlreferences/anti-patterns.mdcli.mdcore-api.mdproviders.mdworkflows.md详细介绍
停止
如果你发现自己导入了以下任一内容,说明走错了路:
- 不要使用 LangChain / LlamaIndex / Haystack / LangGraph。 分块使用
document_splitter。搜索使用.similarity()。工具使用pxt.tools()+invoke_tools()。 - 不要将 pandas 用作工作存储。 表就是存储。
.collect().to_pandas()仅用于导出。 - 不要编写调用模型的
for row in ...:循环。 将调用封装在计算列中。 - 不要安装单独的向量数据库。 在应用中,在模型上设置
__indexes__ = [pxt.EmbeddingIndex(...)]。在 notebook 中,使用t.add_embedding_index(col, embedding=fn)。使用.similarity(string=query)进行搜索。 - 不要编写
while not done:Agent 循环。 插入一行。计算列链会自动运行。
参见 anti-patterns.md(6 个宏)。
什么是 Pixeltable?
一个应用文件(app.py)就是后端。
pxt schema update:根据TableModel类创建表。不会启动 HTTP。- 插入示例数据,使用
.select()、pxt dashboard或pxt schema diff。插入时会运行计算。在执行pxt service update后,使用 curl POST。 pxt service update:启动 HTTP(本地或pxt://)。pxt service list会打印 URL。这是服务命令;不要使用pxt service run。
首次运行:快速入门。了解原因:为什么选择 Pixeltable。
启动新项目
pip install 'pixeltable[serve]' # Python 3.11+
pxt init
pxt service example --out app.py
pxt schema check app.py # validates the file; warns if 'app' is shadowed
pxt schema update app.py my_app
pxt service update app.py my_app -f # no TTY: without -f a pending start exits 3
pxt service list # assigned port; do not hard-code :8000
pxt service example 会写入模型和一个 FastAPIRouter。仅生成架构(不含 HTTP):pxt schema example --brief --out app.py。然后编辑 app.py,再运行 pxt schema update。架构变更后,如果存在路由,请再次运行 pxt service update ... -f;在此之前,它们会返回 409。不要运行 python app.py。完整标志: cli.md。
最后一个参数(my_app,或 Cloud 上的 pxt://org:db)是目录,而不是磁盘上的文件夹。pxt init 会标记项目根目录。本地句柄:pxt.get_table('my_app.docs'),或绑定模型:import app; app.TableModel.bind_all('my_app'),然后使用 app.Docs.insert(...) / app.Docs.select(...).collect()。检查:pxt ls -l、pxt errors my_app/docs、pxt dashboard。
Provider 密钥(OPENAI_API_KEY 等)来自环境变量或 ~/.pixeltable/config.toml,pxt 守护进程会在启动时读取两者一次;本地服务从守护进程获取这些密钥。更改任一项后,运行 pxt daemon restart(在此之前,每条命令都会返回 409),然后运行 pxt service restart my_app/ingest。
同一个文件也可用于 Cloud:优先使用 --add-config-entry,不要手动写入托管的 [[pixeltable.database]] 条目,然后:
pxt login # browser sign-in; or export PIXELTABLE_API_KEY, which wins
pxt db update pxt://org:db --add-config-entry -f # project files, image, workers; not secrets, rows, or HTTP
pxt schema update app.py pxt://org:db -f
pxt service update app.py pxt://org:db -f
Provider 调用会自行重试(速率限制、503、超时,并采用退避策略),并通过 OPENAI_BASE_URL 访问代理:不要为模型调用套上自己的重试循环或客户端(providers.md)。在编码 Agent 环境下,请让用户在自己的终端中运行 pxt login,绝不要在聊天中询问密钥的值。
从 Pixeltable 0.7.20 起,因缺少 -f 或 --allow-destructive 而被拒绝的命令会以状态码 3 退出,并以 next: <command> # <what it does> 结尾;pxt errors 会为每个可重新计算的失败列打印相同内容(快照列不会改变,因此不会提及重新计算)。如果该命令能完成用户要求的操作,就运行它;如果它会删除数据、停止或重启其他人正在调用的服务,或重新计算用户未提及的行,则先询问。当 Pixeltable 无法完成任务所需的操作,或你采用了变通方案时,请向用户展示 pxt feedback --kind gap '<what was needed, what you tried, what happened>';该命令会打印预览,但不会发送任何内容;仅在用户同意后添加 --send。单引号可防止 shell 执行其中的任何文本。使用占位数据,绝不要使用客户名称、主机、路径或密钥。
Cloud 上的 Provider 密钥:pxt secret set pxt://org OPENAI_API_KEY=...,绝不要写入 pixeltable.toml。Cloud 句柄:pxt.get_table('pxt://org:db/docs')。媒体会存储到数据库的托管 home bucket,除非列设置了 destination=;通过链接或 /sign 分享文件(Cloud 参见 references/cli.md)。通过在 dashboard 中插入数据并运行 pxt schema diff 来试用应用。使用 pxt service logs / pxt db logs 查看失败信息。
应用文件
pxt service example --out app.py 会写入这种结构。编辑它,然后运行 pxt schema update app.py my_app。
import pixeltable as pxt
import pixeltable.functions as pxtf
from pixeltable.serving import FastAPIRouter
TableModel = pxt.model_base()
@pxt.udf
def excerpt(text: str, n: int = 12) -> str:
return text if len(text) <= n else f'{text[:n]}...'
class Docs(TableModel, name='docs'):
id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
title: pxt.String
body: pxt.String | None
title_upper = pxtf.string.upper(title)
summary = excerpt(title)
ingest = FastAPIRouter(name='ingest')
ingest.add_insert_route(
Docs, path='/docs', inputs=[Docs.title, Docs.body],
outputs=[Docs.id, Docs.title_upper, Docs.summary],
)
ingest.add_update_route(
Docs, path='/docs/update', inputs=[Docs.title],
outputs=[Docs.id, Docs.title_upper],
)
ingest.add_compute_route(Docs, path='/titles', inputs=[Docs.title], outputs=[Docs.title_upper])
注解是存储列。赋值是计算列。Optional 表示 T | None。主键是 pxt.Column(..., primary_key=True);add_update_route 会按主键匹配行,因此即使 inputs 中未列出 id,请求体仍需携带它。模型上的索引:__indexes__ = [pxt.EmbeddingIndex(...)]。
如果已有 FastAPI:架构更新后,运行 ingest.bind('my_app'),然后运行 app.include_router(ingest)。或者在 app.py 中定义 fastapi.FastAPI 对象,并在那里对每个路由器调用 include_router();随后 pxt service update 会提供该单个应用。请在自定义处理程序中调用 pxt.get_table()。workflows.md。
RAG、视图和搜索:workflows.md。除非用户提出要求,否则不要添加 Hugging Face 或 spaCy。
应用与笔记本
- 应用:
app.py+pxt schema update+pxt service update。索引建在模型上。 - 笔记本 / REPL:
pxt.create_table()、add_computed_column()、add_embedding_index()。下方附录采用这种形式。
查看位置
| 需求 | 打开 |
|---|---|
pxt schema、pxt service、检查 | cli.md |
| 类型、视图、UDF、UDA、工具调用 | core-api.md |
| Provider 导入、参数和输出格式 | providers.md |
| 服务、FastAPIRouter、路由 | workflows.md |
| 技术栈不匹配 | anti-patterns.md |
通过编辑 app.py 添加视频、音频、Agent 或 UI。视图可以是筛选器(base=Docs.where(...)),也可以是迭代器(frame_iterator、audio_splitter、document_splitter、video_splitter、string_splitter、list_iterator、tile_iterator)。编写 UDF 前,先检查 pixeltable.functions。从 pxt service example 或 pxt schema example 开始。不要另造一条 pxt schema update 路径。
API 陷阱
| 错误 | 正确 |
|---|---|
openai.vision(...) | 已弃用(pixeltable.functions 中唯一已弃用的函数)。使用带有 image_url 的 chat_completions,或使用 responses |
from pixeltable.iterators import ... | 整个 pixeltable.iterators 包都是已弃用的兼容垫片(FrameIterator、VideoSplitter、DocumentSplitter、StringSplitter、AudioSplitter、TileIterator)。从 pixeltable.functions.* 导入函数——例如 from pixeltable.functions.video import frame_iterator |
similarity(query) | similarity(string=query)。也可使用 image= / audio= / video= / document= / vector=;idx= 用于在同一列上的多个索引中进行选择 |
使用 if_exists='ignore' 重新运行以修复逻辑 | 应用:编辑表达式,运行 pxt schema update,然后运行 pxt recompute my_app/docs summary -f:更新会保留旧值。Notebook:运行 t.alter_computed_column(summary=...),这会重新计算该列及其依赖项 |
原地更改列的类型(T 改为 T | None、存储列改为计算列、stored=) | UNSUPPORTED:整个更新都不会应用,且 --allow-destructive 无济于事。计算列:重命名(执行一次 --allow-destructive -f)。存储列:重命名会删除其数据,因此应预先声明为 T | None |
t.summary_errortype | 对存储的计算列或媒体列,使用 t.summary.errortype / t.summary.errormsg。媒体列使用 t.<col>.fileurl / .localpath |
pxt.Required[pxt.String] | 默认不可为 null。可选:T | None |
@pxt.udf def f(x: str) 接收可为 null 的列 | 不可为 null 的参数收到 None 时,会跳过调用:单元格为 None,且 errormsg 为空。将 x 标注为 str | None,并在函数体中处理 None |
在 UDF 函数体内调用 whisper.load_model(...) | 权重会在每一行重新加载。使用随附的封装(pxtf.whisper.transcribe、clip.using(...)),或在模块作用域使用缓存的加载器 |
recompute_columns(columns=['summary']) | t.recompute_columns('summary', errors_only=True) |
| TOML 路由或已退役的服务 CLI | FastAPIRouter + pxt schema update + pxt service update |
在 app.py 中调用 add_embedding_index() | 在 TableModel 上使用 __indexes__。注意,DSL 使用 name= 命名索引,SDK 使用 idx_name= |
make_video(order_by=...) / stitch_tiles(order_by=...) | 两者都是 requires_order_by UDA:排序表达式是第一个位置参数——make_video(t.pos, t.frame, fps=25)。使用 order_by= 会报错 |
在 app.py 导入时调用 pxt.create_table() / get_table() | 使用 TableModel + pxt schema update。导入时不得修改目录 |
EmbeddingIndex(frame, image_embed=clip) | 使用 embedding=clip(同时支持文本和图像)。或者同时使用 string_embed= 和 image_embed=。仅使用 image_embed= 无法回答 similarity(string=...) |
uuid.astype(pxt.String) | 使用 uuid.to_string()(from pixeltable.functions.uuid import to_string)。astype 不会将 UUID 转换为 String |
对字典或列表使用 json_col.astype(pxt.String) | 插入时会失败:Expected string, got dict。使用 pxtf.json.dumps(json_col) 进行序列化 |
提取字段(.text、.choices[0].message.content)。该字段是一个存有字符串的 Json:在拼接或生成嵌入前,对它使用 .astype(pxt.String)。
Notebook / REPL 附录
import pixeltable as pxt
pxt.create_dir('my_project', if_exists='ignore')
t = pxt.create_table('my_project.documents', {
'title': pxt.String,
'content': pxt.String,
'image': pxt.Image,
'video': pxt.Video,
'audio': pxt.Audio,
'doc': pxt.Document,
}, if_exists='ignore')
类型默认不可为 null。可选类型为 T | None。不要使用 pxt.Required。
from pixeltable.functions.uuid import uuid7
t = pxt.create_table('my_project.items', {
'content': pxt.String,
'uuid': uuid7(),
}, primary_key=['uuid'], if_exists='ignore')
插入:t.insert([{...}])。计算列:
from pixeltable.functions.openai import chat_completions
t.add_computed_column(
summary=chat_completions(
messages=[{'role': 'user', 'content': t.content}],
model='gpt-4o-mini',
).choices[0].message.content,
if_exists='ignore',
)
视图:document_splitter、frame_iterator(来自 pixeltable.functions.video)、string_splitter、audio_splitter。Notebook 索引:t.add_embedding_index('content', embedding=embed_fn, if_exists='ignore')。
查询:t.where(...).select(...).collect()。相似度:t.content.similarity(string=query)。在 @pxt.query 中,将其别名为 score=sim。
UDF 按相对于项目根目录的模块路径记录(app.excerpt)。
在 Notebook 中调用 create_* / add_* 时,始终使用 if_exists='ignore'。单元格执行失败时:t.recompute_columns('summary', errors_only=True),或执行 pxt recompute my_project/documents summary --errors-only -f。string_splitter / document_splitter(..., separators='sentence') 需要 spaCy。嵌入索引需要使用 .using(...) 来绑定模型参数。