AgentField Python SDK完全教程:用3个核心装饰器把普通函数变成AI Agent API
【免费下载链接】agentfieldBuild, run and scale AI agents like API and microservices项目地址: https://gitcode.com/gh_mirrors/ag/agentfield
AgentField Python SDK 的核心思路是"零胶水代码":你只需要写普通的 Python 函数,再用 3 个核心装饰器@reasoner、@on_event、@on_schedule稍作标记,AgentField 控制平面就会自动把它们变成带路由、队列、重试和可观测性的 AI Agent REST API。本教程面向新手,带你用不到 20 行代码完成从零到上线。
上图展示了 AgentField 的整体架构:你的服务(前端、后端、外部 API)通过 REST/Webhooks 调用 AgentField 控制平面(Control Plane),控制平面再分发到由 Python、Go、TypeScript SDK 驱动的分布式 Agent 节点。所有 Python 编写的 Agent 都运行在sdk/python/agentfield/目录下这个统一的 SDK 之上。
一分钟安装与初始化
安装非常简单,一条命令即可:
pip install agentfield初始化一个 Agent 实例,只需三行:
from agentfield import Agent, AIConfig app = Agent( node_id="hello-world", agentfield_server="http://localhost:8080", # 控制平面地址 ai_config=AIConfig(model="openai/gpt-4o-mini"), )💡
node_id是 Agent 在集群中的唯一标识,后续所有 API 路径(如POST /api/v1/execute/hello-world.say_hello)都由它派生。
完整入门示例可参考 main.py,官方快速上手说明见 sdk/python/README.md。
装饰器①:@reasoner 把普通函数变成 AI Agent API
@reasoner是最核心的装饰器。它把一个async def函数自动注册为可调用的能力(capability),并暴露为 REST 端点。
@app.reasoner(tags=["greeting"]) async def say_hello(name: str) -> dict: """AI 生成一句个性化问候。""" text = await app.ai(user=f"用一句话欢迎 {name} 来到 AgentField") return {"greeting": text}它背后做了 3 件"大事"(实现见 decorators.py):
| 自动能力 | 说明 |
|---|---|
| 📡 自动生成 REST API | 函数签名 + Pydantic 类型注解自动变成请求/响应 Schema |
| 🔍 工作流追踪 | 每次调用自动上报"开始/完成/出错"事件到控制平面 |
| 🛠️ 工具调用发现 | 其他 Agent 的 LLM 可以通过tools="discover"自动发现并调用它 |
你可以为函数添加tags(用于分组与授权)、description(人类可读描述,默认取 docstring)、自定义path等参数,全部通过装饰器参数声明,无需任何路由配置。
装饰器②:@on_event 让 Webhook 事件自动触发 Agent
想让 Stripe 支付成功、GitHub PR 打开这类外部事件自动触发 Agent?只需在@reasoner下面叠加一行@on_event:
@app.reasoner() @on_event( source="stripe", types=["payment_intent.succeeded"], secret_env="STRIPE_SECRET", ) async def handle_payment(input, ctx) -> dict: """收到支付成功事件时自动执行退款/积分逻辑。""" return {"order": input.get("data", {}).get("object", {}).get("id")}关键优势:你不需要自己配置任何 Webhook。Agent 注册时,控制平面会自动为每个事件绑定创建一条 Trigger 记录(类型定义见 triggers.py),并处理签名校验、幂等键、事件重放。每次事件触发后,ctx.trigger里包含事件类型、接收时间等元数据,方便你审计。
装饰器③:@on_schedule 一行代码实现定时任务
周期性 Agent(日报生成、库存巡检、定时备份)用@on_schedule声明即可:
@app.reasoner() @on_schedule("*/5 * * * *") # 每 5 分钟 async def health_check(input, ctx) -> dict: """定时巡检:调用 AI 分析最近日志。""" return {"status": "ok"}@on_schedule等价于ScheduleTrigger(cron=...),支持标准 5 段 cron 表达式和 IANA 时区参数(默认 UTC)。和@on_event一样,定时任务的控制平面调度完全托管——Agent 进程不在线时事件会进入队列,不丢任务。
三件套组合:从函数到生产级 Agent API
回顾一下完整的"装饰器公式":
@reasoner—— 把函数变成可被任何人(服务、其他 Agent、LLM 工具调用)调用的 AI Agent API;@on_event—— 让外部 Webhook 事件自动触发它;@on_schedule—— 让它按 cron 周期自动运行。
最后用一行app.run()启动,控制平面会完成注册、发现、路由。之后你在控制平面 Web UI 的 Executions 页面就能实时看到每一次 Agent 调用的状态、耗时和可验证凭证(VC),真正做到"像调用微服务一样调用 AI Agent"。
🚀 下一步建议:阅读 docs/DEVELOPMENT.md 了解控制平面接入细节,或参考 examples/python_agent_nodes/ 目录下的 RAG、多模态、审批流等完整示例,快速构建你自己的 AI Agent 集群。
【免费下载链接】agentfieldBuild, run and scale AI agents like API and microservices项目地址: https://gitcode.com/gh_mirrors/ag/agentfield
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考