agentl

网站 · 入门 · 连接器 · PyPI · 路线图 · 讨论 · X · Instagram

用 YAML 编写 Agentic ETL。 读取文件、API 和数据库，让大模型根据架构对每条记录进行分类或提取，检查数据并交付。内置逐条记录缓存、预算、重试和实时仪表盘。

• 每条记录只付费一次： 对未更改数据重新运行时不会调用模型；即使作业中途失败，也只需为尚未完成的部分付费（前后对比）。
• 值得信赖的数据： 使用固定规则的 check 步骤，并在交付前进行 Agent 审查。
• 适用于数据所在的各种位置： CSV、Excel、PDF、SQL、Postgres、S3、SharePoint、Google Sheets、Kafka、SQS、webhook，以及适用于 Claude Code 和 Cursor 的 MCP 服务。
• 如实呈现数据： 提供与纯 Python 对比的基准测试和评估，包括表现较差的场景。

状态：测试版（v0.10）。次要版本仍可能更改 API；所有更改都会记录在更新日志中。

面向 AI Agent： 文档位于 agentl.com.br/llms.txt，每个文档页面都提供原始 Markdown 副本（<page>/index.md）。

安装

pip install agentl-etl                 # core: CSV, JSON, HTTP, SQLite, runner, dashboard
pip install 'agentl-etl[office,pdf]'   # Excel, Word, PDF
pip install 'agentl-etl[anthropic]'    # Claude as the LLM provider
pip install 'agentl-etl[all]'

该软件包在 PyPI 上的名称是 agentl-etl；导入名称和 CLI 均为 agentl。

管道

name: invoices
llm:
  provider: anthropic          # or "echo" to run offline
  model: claude-opus-5-5

steps:
  - id: invoices
    use: csv.read
    with: { path: data/invoices.csv }

  - id: customers
    use: json.read
    with: { path: data/customers.json }

  - id: matched
    use: join
    input: [invoices, customers]
    with: { left: invoices, right: customers, on: tax_id, how: left }

  - id: classify
    agent: Classify the expense category and flag unknown clients for review.
    input: matched
    output: { number: str, customer: str?, category: str?, needs_review: bool? }

  - id: save
    use: sqlite.write
    input: classify
    with: { db: out/invoices.db, table: invoices }

notify:
  channels:
    slack: { webhook: "${SLACK_WEBHOOK}" }
  rules:
    - on: run.failed
      to: [slack]
      summary: agent             # the LLM explains the failure in 3 lines
    - on: item.rejected
      when: "rate > 0.05"
      to: [slack]

agentl validate pipeline.yaml
agentl run pipeline.yaml --show classify
agentl diagram pipeline.yaml      # Mermaid
agentl dashboard                  # http://127.0.0.1:7070

也可以编写 extract:、transform: 和 load: 部分来代替 steps:。在 transform 和 load 中，没有 input 的步骤会读取上一步的输出。

概念

一切皆为步骤。 一个步骤从输入中接收记录（list[dict]）并返回记录。Extract、transform 和 load 只是标签，并非不同的处理机制。步骤构成一个图，因此 ETL、ELT、扇出和联接都可以实现。

步骤类型	YAML	用途
连接器	use: pdf.read	读取和写入文件、API 和数据库
函数	fn: my_module:clean	编写自己的确定性代码
Agent	agent: "..." + output:	语义处理：分类、提取、规范化、审计

Agent 步骤会并行地（通过 concurrency）逐条处理记录，并且必须声明 output 架构（str、int、float、bool、list[str]、dict；末尾加上 ? 表示可选）。输出会经过验证；某条记录失败两次后会被拒绝，但不会导致整个运行失败，并会触发 item.rejected 事件。结果会按记录缓存在 .agentl/agentl.db 中，因此崩溃后重新运行时只需为剩余部分付费。

密钥从环境变量中读取：${VAR} 或 ${VAR:-default}。切勿将密钥写入 YAML。

内置连接器

file.read csv.read csv.write json.read json.write txt.read excel.read parquet.read pdf.read docx.read doc.read xml.read html.read http.get http.post sqlite.read sqlite.write select filter dedupe join concat

file.read 会检测格式并解压 ZIP/GZ 压缩包。读取器接受 path（glob）、url（HTTP 下载）或上游记录。agentl connectors 会列出已安装的连接器。http.get 的 URL 中可以使用 {field} 占位符，为每条上游记录各发送一次请求；还可以使用 items: data.results 从响应中选出列表。

扩展

from agentl import connector, register_provider


@connector("crm.contacts")
def contacts(input, since=None):
    return [{"id": 1, "email": "[email protected]"}]

软件包可以通过 agentl.connectors 入口点组注册连接器，因此只需 pip install agentl-yourthing。大模型提供方通过 register_provider(name, factory) 接入；工厂会返回一个包含 structured(instruction, record, schema) 和 text(instruction, content) 的对象。

事件

运行器会发出 run.started、step.started、step.succeeded、step.failed、item.rejected、source.empty、run.succeeded 和 run.failed 事件。这些事件会传递给存储、仪表板和通知规则（slack、webhook、console）。

路线图

接下来将推出 pandas 桥接、支持 upsert 的 Postgres，以及面向 AI Agent 的文档。 查看完整的公开路线图，并为你需要的议题投票。

贡献

欢迎提交错误报告、连接器需求和拉取请求。请阅读 CONTRIBUTING.md， 了解设置方法以及每个 PR 必须通过的检查；如需私下报告漏洞，请阅读 SECURITY.md。 发布历史记录在 CHANGELOG.md 中。

贡献者

感谢所有帮助构建 agentl 的人：

想加入他们吗？带有 good first issue 标签的议题是很好的起点。

赞助商

agentl 免费且采用 MIT 许可证。如果它为你节省了时间，不妨赞助这个项目。

许可证

MIT