12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485 |
- import uuid
- from fastapi import FastAPI
- from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse
- from fastapi.middleware.cors import CORSMiddleware
- from agno.agent import Agent
- from agno.models.openai.like import OpenAILike
- from agno.memory.v2.db.sqlite import SqliteMemoryDb
- from agno.memory.v2.memory import Memory
- from agno.storage.sqlite import SqliteStorage
- from textwrap import dedent
- import os
- memory_db = SqliteMemoryDb(db_file="tmp/chat_memory.db", table_name="memory")
- storge_db = SqliteStorage(table_name="agent_sessions", db_file="tmp/chat_memory.db")
- memory = Memory(
- model=OpenAILike(
- id="qwen3-32b",
- api_key=os.getenv("BAILIAN_API_KEY"),
- base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
- request_params={"extra_body": {"enable_thinking": False}},
- ),
- db=memory_db,
- )
- agent = Agent(
- model=OpenAILike(
- id="qwen3-32b",
- api_key=os.getenv("BAILIAN_API_KEY"),
- base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
- request_params={"extra_body": {"enable_thinking": False}},
- ),
- instructions=dedent("""\
- You are a helpful multi-turn information collection assistant.
- Your job is to collect the user's:
- 1. name
- 2. age
- 3. industry
- You must follow this strict logic:
- - Start by asking the user's name.
- - Then ask for the age.
- - Then ask for the industry.
- - Do NOT skip ahead.
- - If the user gives an irrelevant answer or skips, gently remind them and guide them back.
- - When all 3 pieces of info are collected, show a summary to the user and tell them they are free to ask anything.
- - Until all info is collected, don't answer other questions. Just remind them to finish the info collection first.
- Always remember the information already collected and avoid repeating questions.
- Use concise and friendly tone.
- """),
- memory=memory,
- storage=storge_db,
- stream=True,
- add_datetime_to_instructions=True,
- show_tool_calls=True,
- markdown=False,
- add_history_to_messages=True,
- enable_user_memories=True,
- )
- user_id = str(uuid.uuid4())
- async def ask_agent(message: str):
- for chunk in agent.run(message=message, user_id=user_id, stream=True):
- yield f"data: {chunk}\n\n"
- app = FastAPI()
- @app.get("/stream_text")
- async def stream_text(message: str = "你好,这是使用FastAPI和SSE实现的打字机效果。"):
- return StreamingResponse(
- ask_agent(message),
- media_type="text/event-stream",
- headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
- )
- @app.get("/", response_class=HTMLResponse)
- async def get_index():
- return FileResponse("./index.html")
|