123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110 |
- import asyncio
- from agno.storage.sqlite import SqliteStorage
- from agno.utils.log import logger
- import httpx
- from dotenv import load_dotenv
- from agno.agent import Agent
- from agno.tools import tool
- from typing import Any, Callable, Dict
- from agno.models.openai import OpenAIChat,OpenAILike
- from agno.tools import FunctionCall
- from rich.console import Console
- from rich.pretty import pprint
- from rich.prompt import Prompt
- from agno.exceptions import RetryAgentRun,StopAgentRun,AgentRunException
- from agno.tools.baidusearch import BaiduSearchTools
- from agno.tools import Toolkit
- from agno.memory.v2.memory import Memory
- from agno.memory.v2.db.sqlite import SqliteMemoryDb
- import os
- 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}},
- )
- # 1. 查询联系人
- @tool(
- name="get_contact",
- description="查询用户是否存在,并返回手机号",
- )
- def get_contact(user_name: str):
- # 只有“张三”存在
- if user_name == "张三":
- return {"exists": True, "phone": "13800001111", "message": "用户张三存在,手机号13800001111"}
- else:
- return {"exists": False, "phone": "", "message": f"用户{user_name}不存在,请重新输入正确的姓名"}
- # 2. 查询余额
- @tool(
- name="get_balance",
- description="查询当前账户余额",
- )
- def get_balance(card_number: str):
- return {"balance": 500.0, "message": "当前余额为500元"}
- # 3. 转账
- @tool(
- name="transfer",
- description="向指定用户手机号转账",
- )
- def transfer(user_name: str, phone: str, amount: float,card_number: str):
- if user_name != "张三" or phone != "13800001111" or card_number == None:
- return {"success": False, "message": "收款人信息有误,请重新确认姓名,手机号和卡号"}
- if amount > 500:
- return {"success": False, "message": "余额不足,最多只能转500元"}
- return {"success": True, "message": f"成功向{user_name}({phone})转账{amount}元"}
- # 4. 回复用户(澄清/最终回复)
- @tool(
- name="reply_to_user",
- description="向用户澄清或回复最终结果",
- )
- def reply_to_user(message: str):
- print(f"系统回复用户:{message}")
- return {"ok": True}
- # memory = Memory(db=SqliteMemoryDb(table_name="user_memories", db_file="D:/pythonai/ai_learning/agno_cache/memory.db"))
- agent = Agent(
- model=model,
- tools=[get_contact, get_balance, transfer, reply_to_user],
- instructions=[
- "你是一个银行转账助手,负责协助用户完成转账操作。首先你需要收集用户的信息",
- "请严格按照顺序提问:先问对方姓名,再问对方卡号,再问金额。",
- "1. 先用 get_contact 查询对方是否存在",
- "2. 用 get_balance 查询余额,余额固定500元。",
- "3. 信息齐全后,用 transfer 工具发起转账。",
- "4. 如果余额不足,提示用户最多只能转500元,并让用户重新输入金额。",
- "5. 如果收款人不存在或手机号不对,提示用户重新输入。",
- "6. 每次和用户澄清、最终结果,都用 reply_to_user 工具回复。",
- "7. 只有所有信息都正确且余额充足时,才提示转账成功。",
- "8. 你必须用用户最新提供的信息调用工具。"
- "9. 请使用中文提问或回答"
- ],
- # memory=memory,
- # enable_agentic_memory=True,
- storage=SqliteStorage(table_name="agent_sessions", db_file="D:/pythonai/ai_learning/agno_cache/data.db"),
- session_id="my_session", # 固定session_id
- add_history_to_messages=True,#是否自动把历史对话消息(chat history)添加到每次发送给大模型的 prompt 里。
- num_history_runs=20,#这里设置为 20,表示每次和模型对话时,会把最近的 3 轮(你和 Agent 的来回)历史消息一并发给模型
- markdown=True,
- )
- user_id = "user1"
- def main():
- print("欢迎使用转账助手!输入 exit 退出。")
- while True:
- user_input = input("你:")
- if user_input.strip().lower() in ["exit", "quit"]:
- print("再见!")
- break
- agent.print_response(user_input, user_id=user_id)
- if __name__ == "__main__":
- main()
|