sse_agents_route.py 1.3 KB

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