123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231 |
- import os
- from dotenv import load_dotenv
- from agno.agent import Agent, RunResponseEvent
- from agno.tools import tool
- from agno.memory.v2.db.sqlite import SqliteMemoryDb
- from agno.memory.v2.memory import Memory
- from agno.storage.sqlite import SqliteStorage
- from agno.models.openai.like import OpenAILike
- from typing import Iterator, Optional, Dict, Any
- from enum import Enum
- import json
- import uvicorn
- from fastapi import FastAPI, Request
- from fastapi.responses import StreamingResponse
- from sse_starlette.sse import EventSourceResponse
- import asyncio
- import json
- from typing import AsyncGenerator
- load_dotenv()
- base_url = os.getenv('BAILIAN_API_BASE_URL')
- api_key = os.getenv('BAILIAN_API_KEY')
- SYSTEM_PROMPT = """
- 你是一个信息收集助手。你需要依次收集用户的姓名、年龄和感兴趣的行业。
- 如果用户输入其他无关内容,请提醒用户需要先完成信息收集。
- 开始对话前,请先介绍自己。
-
- 工具使用规则:
- 1. 每当用户提供有效信息时,必须立即使用 add_info 工具保存信息
- 2. 使用 get_info 工具检查特定信息是否已收集
- 3. 使用 list_info 工具确认所有已收集的信息
- 4. 当用户需要重新收集信息时,使用 reset_info 工具清空所有信息,再重新开始收集
- 信息收集策略:
- 1. 使用开放式问题引导用户提供更多信息
- 2. 对模糊的信息进行澄清和确认
- 3. 根据上下文,智能地推断和补充相关信息
- 4. 保持对话的自然性和连贯性
- 5. 务必保证信息的准确性和完整性
- 6. 只有当信息符合需要收集的内容的规则时(例如姓名必须是符合姓名规则的,年龄必须是1~100内的数字等等),才将其记录到已收集信息中
- 7. 如果用户输入年龄时给出的是出生年份,请帮用户计算实际年龄(今年是2025年)。
- 8. 如果用户输入的年龄没有明确的数字,则此输入无效。
- 9. 如果用户一次性将所有信息给出,请将其拆分为单独的字段进行处理,不必重复提问。
- 对话要求:
- 1. 保持友好、专业的语调
- 2. 每次回复要简洁明了
- 3. 每个信息收集成功后,确认用户输入的内容,格式为:"好的,已经收集到您的(此处为收集的项目)是:",并询问下一个问题。
- 4. 当用户询问一些其他问题时,请礼貱地告诉用户,你是一个信息收集助手,不回答其它不相关问题。提醒用户还没收集完成。
- 5. 所有问题回答并收集完毕后,请务必要总结用户的所有信息,并告知用户"信息已收集完毕"。
- 6. 如果没有对话历史,请先介绍自己,再进行询问。
- 7. 如果用户回答与问题无关,请礼貌且友好地提醒用户需要回答相关问题。
- 8. 请不要直接将问题内容直接输出,而是要根据用户的回答进行智能化的处理和回复。
- 注意事项:
- 除非用户输入与信息收集无关,否则一定要调用对应的tools再给予回复。
- 除非用户输入与信息收集无关,否则一定要调用对应的tools再给予回复。
- 除非用户输入与信息收集无关,否则一定要调用对应的tools再给予回复。
- 除非用户输入与信息收集无关,否则一定要调用对应的tools再给予回复。
- 除非用户输入与信息收集无关,否则一定要调用对应的tools再给予回复。
- 除非用户输入与信息收集无关,否则一定要调用对应的tools再给予回复。
- """
- # 初始化空的session状态
- initial_state = {
- "name": None,
- "age": None,
- "industry": None
- }
- class DataField(str, Enum):
- NAME = "name"
- AGE = "age"
- INDUSTRY = "industry"
- @tool(name="add_info", description="添加或修改用户信息")
- def add_info(agent: Agent, field: DataField, value: str) -> str:
- if field == DataField.NAME:
- agent.session_state["name"] = value
- return json.dumps({"result": f"已收集姓名: {value}"})
- elif field == DataField.AGE:
- try:
- age = int(value)
- agent.session_state["age"] = age
- return json.dumps({"result": f"已收集年龄: {age}"})
- except ValueError:
- return json.dumps({"result": "年龄输入无效,请输入一个正确的数字。"})
- elif field == DataField.INDUSTRY:
- agent.session_state["industry"] = value
- return json.dumps({"result": f"已收集感兴趣的行业: {value}"})
- @tool(name="add_all_info", description="一次性添加所有用户信息")
- def add_all_info(agent: Agent, name: str, age: str, industry: str) -> str:
- results = []
- # 姓名
- if name and isinstance(name, str):
- agent.session_state["name"] = name
- results.append(f"已收集姓名: {name}")
- else:
- results.append("姓名输入无效")
- # 年龄
- try:
- age_int = int(age)
- if 1 <= age_int <= 100:
- agent.session_state["age"] = age_int
- results.append(f"已收集年龄: {age_int}")
- else:
- results.append("年龄输入无效")
- except Exception:
- results.append("年龄输入无效")
- # 行业
- if industry and isinstance(industry, str):
- agent.session_state["industry"] = industry
- results.append(f"已收集感兴趣的行业: {industry}")
- else:
- results.append("行业输入无效")
- # 检查是否全部收集
- if all([
- agent.session_state["name"],
- agent.session_state["age"],
- agent.session_state["industry"]
- ]):
- summary = {
- "姓名": agent.session_state["name"],
- "年龄": agent.session_state["age"],
- "感兴趣的行业": agent.session_state["industry"]
- }
- results.append(f"信息已收集完毕: {summary}")
- return json.dumps({"result": results})
- @tool(name="get_info", description="获取用户信息")
- def get_info(agent:Agent, field: DataField) -> str:
- info = agent.session_state.get(field)
- if info is None:
- info = "未收集" if field != DataField.AGE else "未收集"
- return json.dumps({"result": str(info)})
- @tool(name="list_info", description="列出所有用户信息")
- def list_info(agent:Agent) -> str:
- user_info = agent.session_state
- info = {
- "姓名": user_info.get("name") or "未收集",
- "年龄": user_info.get("age") or "未收集",
- "感兴趣的行业": user_info.get("industry") or "未收集"
- }
- return json.dumps({"result": info})
- @tool(name="reset_info", description="重置所有用户信息")
- def reset_info(agent: Agent) -> str:
- agent.session_state["name"] = None
- agent.session_state["age"] = None
- agent.session_state["industry"] = None
- return json.dumps({"result": "所有信息已重置"})
- # 配置
- user_id = "test"
- session_id = "325"
- db_file = "tmp/agent.db"
- # 模型配置
- model = OpenAILike(
- id="qwen3-32b",
- api_key=api_key,
- base_url=base_url,
- request_params={
- "extra_body": {"enable_thinking": False},
- # "response_format": {"type": "json_object"}
- },
- )
- # 初始化内存和存储
- memory = Memory(
- model=model,
- db=SqliteMemoryDb(table_name="user_memories", db_file=db_file),
- )
- storage = SqliteStorage(table_name="agent_sessions", db_file=db_file)
- # 创建agent实例
- agent = Agent(
- model=model,
- tools=[add_info, add_all_info, get_info, list_info, reset_info],
- instructions=SYSTEM_PROMPT,
- add_datetime_to_instructions=True,
- # show_tool_calls=True,
- markdown=True,
- memory=memory,
- enable_user_memories=True,
- storage=storage,
- add_history_to_messages=True,
- enable_session_summaries=True,
- session_state=initial_state,
- add_state_in_messages=True,
- session_id=session_id,
- )
- app = FastAPI()
- async def generate_agent_response(prompt: str, user_id: str) -> AsyncGenerator[str, None]:
- """生成Agent响应的异步生成器"""
- try:
- response_generator: Iterator[RunResponseEvent] = agent.run(prompt, user_id=user_id, stream=True)
- for chunk in response_generator:
- # RunResponseEvent通常是一个对象,需根据其属性获取内容
- if hasattr(chunk, "content"):
- yield json.dumps({"type": "message", "content": chunk.content})
- else:
- yield json.dumps({"type": "message", "content": str(chunk)})
- # await asyncio.sleep(0.05)
-
- # 发送当前状态
- state_info = json.loads(list_info(agent))
- yield json.dumps({"type": "state", "content": state_info["result"]})
-
- except Exception as e:
- yield json.dumps({"type": "error", "content": str(e)})
- @app.get("/")
- async def root():
- return {"message": "信息收集助手 API"}
- @app.get("/chat/sse")
- async def chat_sse(prompt: str, user_id: str = "default"):
- """SSE聊天端点"""
- return EventSourceResponse(
- generate_agent_response(prompt, user_id),
- media_type="text/event-stream"
- )
- if __name__ == "__main__":
- uvicorn.run(app, host="localhost", port=8000)
|