mcpField
P
pixeltable

pixeltable

github.com/pixeltable开发644 次安装1 天前更新
代码仓库

关于 pixeltable

pixeltable 是面向构建数据密集型 AI 应用的 Python 框架,将表、计算列、媒体处理、向量索引和服务集中在应用文件中。该技能提供项目构建与命令使用指导,并强调以表格计算流程代替外部向量数据库和逐行模型循环。

主要功能

  1. 01以 TableModel 定义表结构
  2. 02插入时运行计算列链
  3. 03支持媒体拆分与处理
  4. 04通过 EmbeddingIndex 建立索引
  5. 05提供 HTTP 服务与命令行流程

适用场景

  1. 01构建基于文档的检索应用
  2. 02在表格中处理媒体与模型输出
  3. 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.yaml
references/
anti-patterns.md
cli.md
core-api.md
providers.md
workflows.md

详细介绍

停止

如果你发现自己导入了以下任一内容,说明走错了路:

  1. 不要使用 LangChain / LlamaIndex / Haystack / LangGraph。 分块使用 document_splitter。搜索使用 .similarity()。工具使用 pxt.tools() + invoke_tools()。
  2. 不要将 pandas 用作工作存储。 表就是存储。.collect().to_pandas() 仅用于导出。
  3. 不要编写调用模型的 for row in ...: 循环。 将调用封装在计算列中。
  4. 不要安装单独的向量数据库。 在应用中,在模型上设置 __indexes__ = [pxt.EmbeddingIndex(...)]。在 notebook 中,使用 t.add_embedding_index(col, embedding=fn)。使用 .similarity(string=query) 进行搜索。
  5. 不要编写 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 路由或已退役的服务 CLIFastAPIRouter + 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(...) 来绑定模型参数。

资源

评论

支持 Markdown