sse_agents_route.py 1.3 KB

123456789101112131415161718192021222324252627282930313233343536
  1. from fastapi import APIRouter, Query
  2. from fastapi.responses import StreamingResponse
  3. import sys
  4. import os
  5. sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  6. from agents.simple_agent import simple_info_agent # 导入基于agno框架的信息收集Agent
  7. agents_route = APIRouter(prefix="/agents",tags=["agents"])
  8. @agents_route.get("/stream")
  9. async def chat_stream(
  10. username: str = Query(..., description="用户名"),
  11. user_id: str = Query(..., description="用户ID"),
  12. user_message: str = Query(..., description="用户消息")
  13. ):
  14. """
  15. SSE接口,流式输出Agent响应。
  16. 用户输入通过GET的query参数传递,包含username、user_id、user_message。
  17. """
  18. if not user_message:
  19. user_message = "你好,请介绍一下你自己。"
  20. def agno_stream():
  21. """
  22. 生成SSE数据流,逐步输出Agent的响应内容。
  23. agno的Agent.run支持stream=True,返回事件生成器。
  24. """
  25. for event in simple_info_agent.run(user_message, stream=True,username=username,user_id=user_id):
  26. if hasattr(event, "content") and event.content:
  27. yield f"data: {event.content}\n\n"
  28. return StreamingResponse(agno_stream(), media_type="text/event-stream")