1.7 KiB
1.7 KiB
可运行示例:独立函数(无服务器)
当应用是一个纯 Python 函数或模块时——没有 Web 框架,没有服务器,没有基础设施。
方法:直接从 run() 中导入并调用该函数。这是最简单的情况。
# pixie_qa/run_app.py
from pydantic import BaseModel
import pixie
class AppArgs(BaseModel):
question: str
class AppRunnable(pixie.Runnable[AppArgs]):
"""驱动一个独立函数,用于追踪和评估。"""
@classmethod
def create(cls) -> "AppRunnable":
return cls()
async def run(self, args: AppArgs) -> None:
from myapp.agent import answer_question
await answer_question(args.question)
如果函数是同步的,用 asyncio.to_thread 包裹它:
import asyncio
async def run(self, args: AppArgs) -> None:
from myapp.agent import answer_question
await asyncio.to_thread(answer_question, args.question)
如果函数依赖外部服务(例如向量存储),你在步骤 2a 中添加的 wrap(purpose="input") 调用会自动处理——注册中心会在评估模式下注入测试数据。
何时使用 setup() / teardown()
大多数独立函数不需要生命周期方法。仅当函数需要共享资源(例如预加载的嵌入模型、数据库连接)时才使用它们:
class AppRunnable(pixie.Runnable[AppArgs]):
_model: SomeModel
@classmethod
def create(cls) -> "AppRunnable":
return cls()
async def setup(self) -> None:
from myapp.models import load_model
self._model = load_model()
async def run(self, args: AppArgs) -> None:
from myapp.agent import answer_question
await answer_question(args.question, model=self._model)