# main.py from fastapi import FastAPI, Request from agno.storage.sqlite import SqliteStorage from sse_starlette.sse import EventSourceResponse from agno.agent import Agent from agno.models.openai import OpenAILike from dotenv import load_dotenv import os app = FastAPI() load_dotenv() # 读取环境变量 model = OpenAILike( id="qwen3-32b", api_key=os.getenv("BAILIAN_API_KEY"), base_url=os.getenv("BAILIAN_API_BASE_URL"), # request_params={"extra_body": {"enable_thinking": False}},#如果不使用流式输出记得加这句话不然会报错 ) # 初始化 Agent agent = Agent( model=model, instructions=( "你是一个信息收集助手,需要和用户多轮对话,收集用户的姓名、年龄和感兴趣的行业。" "请严格按照顺序提问:先问姓名,再问年龄,再问行业。" "每次只问一个问题,问完问题需要加一个?。" "用户如果输入无关内容,请提醒用户先完成信息收集,不要回答无关问题。" "用户如果拒绝回答(如‘不告诉你’),请礼貌提醒必须填写。" "收集完毕后请总结结果。" ), storage=SqliteStorage(table_name="agent_sessions", db_file="D:/pythonai/ai_learning/agno_cache/data.db"), add_history_to_messages=True,#是否自动把历史对话消息(chat history)添加到每次发送给大模型的 prompt 里。 num_history_runs=20,#这里设置为 20,表示每次和模型对话时,会把最近的 20 轮(你和 Agent 的来回)历史消息一并发给模型 ) @app.get("/stream") async def stream(request: Request, prompt: str,session_id:str): async def event_generator(): for chunk in agent.run(prompt, stream=True,session_id=session_id): if await request.is_disconnected(): break if hasattr(chunk, "content") and chunk.content: yield {"data": chunk.content} return EventSourceResponse(event_generator())