FastAPIHelper 集成

适用版本：4.0.8+

FastAPIHelper 是 Agently 提供的 FastAPI 集成层，用于把响应提供者（Agent / Request / Flow）直接转换成 HTTP 接口。

适合场景：

• 需要快速搭建 AI 服务 API
• 同时提供普通请求与流式响应（SSE/WS）
• 想复用 Agently 的输入构建、输出解析和错误包装

1. 支持的 response_provider 类型

FastAPIHelper(response_provider=...) 支持：

• BaseAgent
• ModelRequest
• TriggerFlow
• TriggerFlowExecution
• 自定义 Generator / AsyncGenerator 函数

这使你可以按工程阶段渐进演进：

• 先用 Agent 快速上线
• 再迁移到 TriggerFlow 编排
• 保持统一 API 面向客户端

2. 完整服务示例（含启动）

import os
import uvicorn

from agently import Agently
from agently.integrations.fastapi import FastAPIHelper

OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434/v1")

Agently.set_settings(
    "OpenAICompatible",
    {
        "base_url": OLLAMA_BASE_URL,
        "model": "qwen2.5:7b",
        "options": {"temperature": 0.3},
    },
)

agent = Agently.create_agent()
agent.role("You are a concise and helpful assistant.", always=True)

app = FastAPIHelper(response_provider=agent)

# 一次性挂载 4 种接口
(
    app
    .use_post("/agent/chat")
    .use_get("/agent/chat/get")
    .use_sse("/agent/chat/sse")
    .use_websocket("/agent/chat/ws")
)

@app.get("/health")
async def health():
    return {"ok": True, "provider": "agent", "model": "qwen2.5:7b"}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

3. 请求与响应协议

3.1 请求体结构

统一结构：

{
  "data": { "input": "hello" },
  "options": {}
}

• data：传给 provider 的输入数据
• options：传给 provider 方法的可选参数

3.2 默认响应包装

成功：

{ "status": 200, "data": "...", "msg": null }

失败：

{
  "status": 422,
  "data": null,
  "msg": "Invalid request payload ...",
  "error": {
    "type": "ValueError",
    "module": "builtins",
    "message": "..."
  }
}

默认状态码映射：

• ValueError -> 422
• TimeoutError -> 504
• 其他异常 -> 400

如需自定义格式，可传 response_warper。

4. 各路由使用方式

方法	路径示例	说明
POST	/agent/chat	JSON body 输入，最通用
GET	/agent/chat/get	payload query 参数传 JSON 字符串
SSE	/agent/chat/sse	按 data: 持续推送流事件
WS	/agent/chat/ws	双向实时通信

5. cURL 调试示例

5.1 POST

curl -sS -X POST "http://127.0.0.1:8000/agent/chat" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {"input": "用一句话介绍 Agently"},
    "options": {}
  }'

5.2 SSE

curl -N -G "http://127.0.0.1:8000/agent/chat/sse" \
  --data-urlencode 'payload={"data":{"input":"请从 1 数到 3"},"options":{}}'

6. TriggerFlow 并发联动

当 provider 是 TriggerFlowExecution 且 options 带 concurrency 时， FastAPIHelper 会调用 execution.set_concurrency(...)，便于按请求动态调节并发上限。

{
  "data": {"input": "..."},
  "options": {"concurrency": 2}
}

7. 生产建议

• 在网关层统一鉴权、限流和审计
• 对 payload 进行 schema 校验（FastAPI/Pydantic）
• SSE/WS 场景配置超时与断线重连策略
• 对异常响应保留 error.type 与请求 trace id，便于排障
• 结合 TriggerFlow 将长流程拆分为可观察的 runtime stream

8. 参考示例

• examples/fastapi/fastapi_helper_agent_ollama.py
• examples/fastapi/fastapi_helper_triggerflow_ollama.py
• examples/fastapi/fastapi_helper_agent_request.py