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")