FastAPIHelper 集成

适用版本：4.0.8.1+

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

1. response_provider 到接口协议的映射

如何阅读这张图

• FastAPIHelper 的角色是协议适配层，不是新的 runtime。
• 它把同一个响应提供者包装成多种 HTTP/实时接口，而不是重新定义一套编排系统。

架构思路

这也是为什么你可以先从 Agent 快速上线，再换成 TriggerFlowExecution，而对外 API 形态基本不变。

适合场景：

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

2. 支持的 response_provider 类型

FastAPIHelper(response_provider=...) 支持：

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

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

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)

(
    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)

4. 请求与响应协议

4.1 请求体结构

统一结构：

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

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

4.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。

5. 各路由使用方式

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

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