|
@@ -0,0 +1,85 @@
|
|
|
+
|
|
|
+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")
|