chat.py 2.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485
  1. import uuid
  2. from fastapi import FastAPI
  3. from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse
  4. from fastapi.middleware.cors import CORSMiddleware
  5. from agno.agent import Agent
  6. from agno.models.openai.like import OpenAILike
  7. from agno.memory.v2.db.sqlite import SqliteMemoryDb
  8. from agno.memory.v2.memory import Memory
  9. from agno.storage.sqlite import SqliteStorage
  10. from textwrap import dedent
  11. import os
  12. memory_db = SqliteMemoryDb(db_file="tmp/chat_memory.db", table_name="memory")
  13. storge_db = SqliteStorage(table_name="agent_sessions", db_file="tmp/chat_memory.db")
  14. memory = Memory(
  15. model=OpenAILike(
  16. id="qwen3-32b",
  17. api_key=os.getenv("BAILIAN_API_KEY"),
  18. base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
  19. request_params={"extra_body": {"enable_thinking": False}},
  20. ),
  21. db=memory_db,
  22. )
  23. agent = Agent(
  24. model=OpenAILike(
  25. id="qwen3-32b",
  26. api_key=os.getenv("BAILIAN_API_KEY"),
  27. base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
  28. request_params={"extra_body": {"enable_thinking": False}},
  29. ),
  30. instructions=dedent("""\
  31. You are a helpful multi-turn information collection assistant.
  32. Your job is to collect the user's:
  33. 1. name
  34. 2. age
  35. 3. industry
  36. You must follow this strict logic:
  37. - Start by asking the user's name.
  38. - Then ask for the age.
  39. - Then ask for the industry.
  40. - Do NOT skip ahead.
  41. - If the user gives an irrelevant answer or skips, gently remind them and guide them back.
  42. - When all 3 pieces of info are collected, show a summary to the user and tell them they are free to ask anything.
  43. - Until all info is collected, don't answer other questions. Just remind them to finish the info collection first.
  44. Always remember the information already collected and avoid repeating questions.
  45. Use concise and friendly tone.
  46. """),
  47. memory=memory,
  48. storage=storge_db,
  49. stream=True,
  50. add_datetime_to_instructions=True,
  51. show_tool_calls=True,
  52. markdown=False,
  53. add_history_to_messages=True,
  54. enable_user_memories=True,
  55. )
  56. user_id = str(uuid.uuid4())
  57. async def ask_agent(message: str):
  58. for chunk in agent.run(message=message, user_id=user_id, stream=True):
  59. yield f"data: {chunk}\n\n"
  60. app = FastAPI()
  61. @app.get("/stream_text")
  62. async def stream_text(message: str = "你好,这是使用FastAPI和SSE实现的打字机效果。"):
  63. return StreamingResponse(
  64. ask_agent(message),
  65. media_type="text/event-stream",
  66. headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
  67. )
  68. @app.get("/", response_class=HTMLResponse)
  69. async def get_index():
  70. return FileResponse("./index.html")