from logging import log from textwrap import dedent from agno.agent import Agent from agno.models.openai import OpenAILike from agno.playground import Playground from agno.storage.sqlite import SqliteStorage from agno.tools.yfinance import YFinanceTools from agno.tools import tool import os from agno.utils.log import log_info from dotenv import load_dotenv load_dotenv() #数据存储位置 agent_storage: str = "./曹航/3/transfer/tmp/agents.db" #测试 finance_agent = Agent( name="Finance Agent", 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}},), tools=[YFinanceTools(stock_price=True, analyst_recommendations=True, company_info=True, company_news=True)], instructions=["Always use tables to display data"], storage=SqliteStorage(table_name="finance_agent", db_file=agent_storage), add_datetime_to_instructions=True, add_history_to_messages=True, num_history_responses=5, markdown=True, ) # 从transfer/prompt目录读取prompt文件 # with open("transfer/prompt/prompt.txt", "r", encoding="utf-8") as f: # prompt_text = f.read() prompt_text=""" ## 背景与目标 你是一个金融助手 Agent,使用 Qwen3-32B 模型驱动,负责协助用户完成向他人账户转账的流程。/ 你需要收集用户转账<对象的名称>和<转账的金额>/ 收集到任一项信息后,确认信息是否可用,如果不可用,通过方法告知给用户。/ 当用户提问你其它与转账不相关的问题时,你应该时刻提醒并保持礼貌。/ 你将通过自然语言与用户对话,收集必要信息,完成转账操作。/ 请根据以下流程与规则进行对话与操作。 ## 任务流程 ### 进行信息收集 请通过对话与用户交互,依次确认并记录以下三项转账信息: 1. 用户提供转账账户后,立刻通过对方姓名查询账户是否存在(`get_contact`) 2. 用户提供转账金额后,立刻确认转账金额是否大于账户余额(`get_balance`) 3. 余额不足时,请重新向用户收集信息,继续转账(`replay_to_user`) > 请在用户表达不清或信息缺失时,请通过方法 `replay_to_user` 发起澄清。 ### 执行转账 确认收集信息后,请总结信息,并请发起转账(`transfer`) """ # 创建用户列表 users = [ {"name": "张三","amount": 1000.0}, {"name": "李四","amount": 500.0}, ] # 定义转账相关的工具函数 def get_contact(user_name: str) -> bool: """ 检查账户是否存在 Args: user_name: 账号 Returns: bool: 账户是否存在 """ # 这里模拟账户检查逻辑 if user_name == "张三": return True else: return False def get_balance() -> float: """ 查询账户余额 Returns: float: 账户余额 """ balance=users[1]["amount"] return balance #TODO 做用户确认 # @tool(requires_user_input=True,user_input_fields=["amount"]) def transfer(user_name, phone: str, amount: float) -> bool: """ 执行转账操作 Args: user_name: 转入账号 amount: 转账金额 Returns: bool: 转账是否成功 """ # 这里模拟转账逻辑 users[0]["amount"] += amount print(f"向{user_name}转账{amount}元成功") return True def replay_to_user(message: str)->str: """ 当前步骤需要用户输入时,请通过此方法向用户提问 向用户提问获取信息,并记录到log Args: message: 提问内容 """ # 这里实现向用户提问的逻辑 #只是记录log log_info(message) return "请等待用户回答" transfer_agent = Agent( name="Transfer Agent", 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}},), tools=[get_contact,get_balance,transfer,replay_to_user], #TODO memory记忆 # memory=SqliteStorage(table_name="transfer_agent", db_file=agent_storage), instructions=dedent(prompt_text), # Store the agent sessions in a sqlite database storage=SqliteStorage(table_name="transfer_agent_session", db_file=agent_storage), show_tool_calls=True, # Adds the current date and time to the instructions add_datetime_to_instructions=True, # Adds the history of the conversation to the messages add_history_to_messages=True, # Number of history responses to add to the messages num_history_responses=5, # Adds markdown formatting to the messages markdown=True, ) # 启动playground-server playground = Playground(agents=[transfer_agent, finance_agent]) app = playground.get_app() if __name__ == "__main__": playground.serve("agno_transfer_agent:app", reload=True)