Agno AgentOS FastAPI 服务化架构
Agno AgentOS FastAPI 服务化架构
学习目标
本章将分析 Agno 的生产服务化架构：
- AgentOS 的 FastAPI 应用结构与 50+ 端点
- SSE（Server-Sent Events）和 WebSocket 双通道流式
- JWT RBAC 权限控制与多租户隔离
- 中间件管线与路由系统
- 多渠道接口（Slack、Telegram、WhatsApp、Discord）
项目实践
AgentOS 应用结构
os/app.py（1600+ 行）是 AgentOS 的核心 FastAPI 应用：
# 伪代码：AgentOS FastAPI 应用class AgentOS: def __init__(self, agents: List[Agent], teams: List[Team], workflows: List[Workflow]): self.app = FastAPI() self.registry = Registry()
# 注册路由 self._register_agent_routes() self._register_team_routes() self._register_workflow_routes() self._register_health_routes()
# 注册中间件 self._register_middleware()50+ 端点分类
| 类别 | 端点数量 | 示例 |
|---|---|---|
| Agent | ~15 | POST /v1/agent/run, POST /v1/agent/run/stream |
| Team | ~10 | POST /v1/team/run, POST /v1/team/run/stream |
| Workflow | ~10 | POST /v1/workflow/run, POST /v1/workflow/run/stream |
| Knowledge | ~5 | POST /v1/knowledge/search, POST /v1/knowledge/load |
| Memory | ~5 | GET /v1/memory/{user_id}, DELETE /v1/memory/{user_id} |
| Session | ~5 | GET /v1/sessions/, DELETE /v1/sessions/{session_id} |
| 管理 | ~5 | GET /health, GET /v1/stats |
SSE 与 WebSocket 双通道
# 伪代码：SSE 端点@app.post("/v1/agent/run/stream")async def run_agent_stream(request: AgentRunRequest): async def event_generator(): async for event in agent.arun_stream(...): yield format_sse(event) return EventSourceResponse(event_generator())
# 伪代码：WebSocket 端点@app.websocket("/ws/agent")async def ws_agent(websocket: WebSocket): await websocket.accept() async for message in websocket.iter_text(): async for event in agent.arun_stream(...): await websocket.send_text(serialize(event))SSE 优势：单向流式（服务端 → 客户端），兼容性好，适合网页实时渲染 WebSocket 优势：双向通信，适合交互式场景（客户端可以发送中断、审批等操作）
JWT RBAC 权限控制
# 伪代码：JWT 认证中间件@app.middleware("http")async def auth_middleware(request: Request, call_next): if request.url.path.startswith("/health"): return await call_next(request)
token = request.headers.get("Authorization") if not token: return JSONResponse(status_code=401, content={"error": "Unauthorized"})
claims = verify_jwt(token) request.state.user_id = claims.get("sub") request.state.scopes = claims.get("scopes", []) return await call_next(request)RBAC 范围：
agent:run— 执行 Agentagent:read— 读取 Agent 状态session:delete— 删除会话admin:*— 管理员权限
多租户隔离
os/scopes.py 定义了多租户的权限范围系统：
# 伪代码：多租户范围class Scope: def __init__(self, user_id: str, team_id: Optional[str] = None): self.user_id = user_id self.team_id = team_id
def can_access(self, resource: Resource) -> bool: return ( resource.user_id == self.user_id and (self.team_id is None or resource.team_id == self.team_id) )多渠道接口
os/interfaces/ 目录包含多渠道接口实现：
| 渠道 | 文件 | 功能 |
|---|---|---|
| Slack | slack/ | Slack Bot 集成，接收消息 → Agent → 回复 |
| Telegram | telegram/ | Telegram Bot，类似 Slack |
whatsapp/ | WhatsApp Business API 集成 | |
| Discord | discord/ | Discord Bot 集成 |
| AG-UI | ag_ui/ | Agent UI 协议 |
| A2A | a2a/ | Agent-to-Agent 协议 |
可观测性
AgentOS 集成 OpenTelemetry Tracing：
# 伪代码：OpenTelemetry 集成from opentelemetry import trace
tracer = trace.get_tracer("agno.os")
@app.post("/v1/agent/run")async def run_agent(request: AgentRunRequest): with tracer.start_as_current_span("agent.run") as span: span.set_attribute("agent.id", agent.id) span.set_attribute("user.id", request.user_id) response = await agent.arun(...) span.set_attribute("run.id", response.run_id) return response问题与规避
1. 50+ 端点的维护负担
每个 Agent/Team/Workflow 操作都有 sync 和 async 两个端点。
规避：使用 FastAPI 的 APIRouter 分组，每个模块一个 Router。新增端点时参照已有模块的模式。
2. SSE 连接泄漏
客户端断开连接但服务端的 generator 仍在运行。
规避：
- SSE generator 检测
request.is_disconnected()并提前退出 - WebSocket 在断开时清理关联任务
3. JWT 验证的性能开销
每个请求都验证 JWT token。
规避：
- 使用缓存（如
lru_cache）缓存最近验证的 token /health端点跳过认证- 批量操作可以复用认证结果
设计取舍
为什么同时支持 SSE 和 WebSocket？
SSE：适合网页端实时展示 Agent 响应（单向流式） WebSocket：适合需要双向交互的场景（中断、审批、多轮对话） 替代方案：只支持一种——但不同场景需要不同的协议
为什么用 FastAPI 而非其他框架？
优势：FastAPI 原生支持 async、自动 OpenAPI 文档、Pydantic 集成 替代方案：Flask（不支持 async）、Starlette（FastAPI 的底层，但缺少自动文档）