wehub-skill-sync
60 lines
1.7 KiB
Markdown
60 lines
1.7 KiB
Markdown
# 可运行示例：独立函数（无服务器）
|
|
|
|
**当应用是一个纯 Python 函数或模块时**——没有 Web 框架，没有服务器，没有基础设施。
|
|
|
|
**方法**：直接从 `run()` 中导入并调用该函数。这是最简单的情况。
|
|
|
|
```python
|
|
# 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` 包裹它：
|
|
|
|
```python
|
|
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()`
|
|
|
|
大多数独立函数不需要生命周期方法。仅当函数需要共享资源（例如预加载的嵌入模型、数据库连接）时才使用它们：
|
|
|
|
```python
|
|
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)
|