test3.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. import asyncio
  2. from agno.storage.sqlite import SqliteStorage
  3. from agno.utils.log import logger
  4. import httpx
  5. from dotenv import load_dotenv
  6. from agno.agent import Agent
  7. from agno.tools import tool
  8. from typing import Any, Callable, Dict
  9. from agno.models.openai import OpenAIChat,OpenAILike
  10. from agno.tools import FunctionCall
  11. from rich.console import Console
  12. from rich.pretty import pprint
  13. from rich.prompt import Prompt
  14. from agno.exceptions import RetryAgentRun,StopAgentRun,AgentRunException
  15. from agno.tools.baidusearch import BaiduSearchTools
  16. from agno.tools import Toolkit
  17. from agno.memory.v2.memory import Memory
  18. from agno.memory.v2.db.sqlite import SqliteMemoryDb
  19. import os
  20. load_dotenv()
  21. # 读取环境变量
  22. model = OpenAILike(
  23. id="qwen3-32b",
  24. api_key=os.getenv("BAILIAN_API_KEY"),
  25. base_url=os.getenv("BAILIAN_API_BASE_URL"),
  26. request_params={"extra_body": {"enable_thinking": False}},
  27. )
  28. # 1. 查询联系人
  29. @tool(
  30. name="get_contact",
  31. description="查询用户是否存在,并返回手机号",
  32. )
  33. def get_contact(user_name: str):
  34. # 只有“张三”存在
  35. if user_name == "张三":
  36. return {"exists": True, "phone": "13800001111", "message": "用户张三存在,手机号13800001111"}
  37. else:
  38. return {"exists": False, "phone": "", "message": f"用户{user_name}不存在,请重新输入正确的姓名"}
  39. # 2. 查询余额
  40. @tool(
  41. name="get_balance",
  42. description="查询当前账户余额",
  43. )
  44. def get_balance(card_number: str):
  45. return {"balance": 500.0, "message": "当前余额为500元"}
  46. # 3. 转账
  47. @tool(
  48. name="transfer",
  49. description="向指定用户手机号转账",
  50. )
  51. def transfer(user_name: str, phone: str, amount: float,card_number: str):
  52. if user_name != "张三" or phone != "13800001111" or card_number == None:
  53. return {"success": False, "message": "收款人信息有误,请重新确认姓名,手机号和卡号"}
  54. if amount > 500:
  55. return {"success": False, "message": "余额不足,最多只能转500元"}
  56. return {"success": True, "message": f"成功向{user_name}({phone})转账{amount}元"}
  57. # 4. 回复用户(澄清/最终回复)
  58. @tool(
  59. name="reply_to_user",
  60. description="向用户澄清或回复最终结果",
  61. )
  62. def reply_to_user(message: str):
  63. print(f"系统回复用户:{message}")
  64. return {"ok": True}
  65. # memory = Memory(db=SqliteMemoryDb(table_name="user_memories", db_file="D:/pythonai/ai_learning/agno_cache/memory.db"))
  66. agent = Agent(
  67. model=model,
  68. tools=[get_contact, get_balance, transfer, reply_to_user],
  69. instructions=[
  70. "你是一个银行转账助手,负责协助用户完成转账操作。首先你需要收集用户的信息",
  71. "请严格按照顺序提问:先问对方姓名,再问对方卡号,再问金额。",
  72. "1. 先用 get_contact 查询对方是否存在",
  73. "2. 用 get_balance 查询余额,余额固定500元。",
  74. "3. 信息齐全后,用 transfer 工具发起转账。",
  75. "4. 如果余额不足,提示用户最多只能转500元,并让用户重新输入金额。",
  76. "5. 如果收款人不存在或手机号不对,提示用户重新输入。",
  77. "6. 每次和用户澄清、最终结果,都用 reply_to_user 工具回复。",
  78. "7. 只有所有信息都正确且余额充足时,才提示转账成功。",
  79. "8. 你必须用用户最新提供的信息调用工具。"
  80. "9. 请使用中文提问或回答"
  81. ],
  82. # memory=memory,
  83. # enable_agentic_memory=True,
  84. storage=SqliteStorage(table_name="agent_sessions", db_file="D:/pythonai/ai_learning/agno_cache/data.db"),
  85. session_id="my_session", # 固定session_id
  86. add_history_to_messages=True,#是否自动把历史对话消息(chat history)添加到每次发送给大模型的 prompt 里。
  87. num_history_runs=20,#这里设置为 20,表示每次和模型对话时,会把最近的 3 轮(你和 Agent 的来回)历史消息一并发给模型
  88. markdown=True,
  89. )
  90. user_id = "user1"
  91. def main():
  92. print("欢迎使用转账助手!输入 exit 退出。")
  93. while True:
  94. user_input = input("你:")
  95. if user_input.strip().lower() in ["exit", "quit"]:
  96. print("再见!")
  97. break
  98. agent.print_response(user_input, user_id=user_id)
  99. if __name__ == "__main__":
  100. main()