#!/usr/bin/env python3 import os import sys from typing import Iterator import dotenv from agno.agent import Agent from agno.memory.v2.db.sqlite import SqliteMemoryDb from agno.memory.v2.memory import Memory from agno.models.openai import OpenAILike from agno.storage.sqlite import SqliteStorage from agno.tools.yfinance import YFinanceTools # 加载环境变量 dotenv.load_dotenv() # 全局联系人数组 CONTACTS = [ {"name": "张三", "phone": "138-0000-1001"}, {"name": "李四", "phone": "139-0000-1002"}, {"name": "王五", "phone": "136-0000-1003"}, {"name": "赵六", "phone": "137-0000-1004"}, {"name": "孙七", "phone": "135-0000-1005"}, {"name": "周八", "phone": "133-0000-1006"}, {"name": "吴九", "phone": "188-0000-1007"}, {"name": "郑十", "phone": "180-0000-1008"} ] # 转账工具 def transfer_money(name: str, phone: str, amount: float): """转账工具""" return f"转账成功,转账金额为:{amount},转账给:{name},转账电话:{phone}" def find_contact_by_name(name: str): """根据姓名查找联系人信息""" for contact in CONTACTS: if name in contact["name"] or contact["name"] in name: return f"找到联系人:{contact['name']},电话:{contact['phone']}" return f"未找到姓名包含'{name}'的联系人" def get_account_balance(): """随机生成账户余额""" import random # 生成1000到50000之间的随机余额 balance = round(random.uniform(1000, 50000), 2) return f"您的当前账户余额为:¥{balance:,.2f}" def create_memory_agent(): """创建具有Memory功能的Agent""" # 检查环境变量 api_key = os.getenv("BAILIAN_API_KEY") base_url = os.getenv("BAILIAN_API_BASE_URL") if not api_key or not base_url: print("❌ 错误: 请确保设置了以下环境变量:") print(" - BAILIAN_API_KEY") print(" - BAILIAN_API_BASE_URL") print("\n💡 您可以创建 .env 文件来设置这些变量") return None try: print("🚀 正在创建具有Memory功能的Agent...") # 创建模型 model = OpenAILike( id="qwen3-32b", api_key=api_key, base_url=base_url, request_params={"extra_body": {"enable_thinking": False}}, ) # UserId for the memories user_id = "user_001" # Database file for memory and storage db_file = "tmp/agent_memory.db" # 创建tmp目录(如果不存在) os.makedirs("tmp", exist_ok=True) # Initialize memory.v2 memory = Memory( model=model, # 使用相同的模型进行记忆管理 db=SqliteMemoryDb(table_name="user_memories", db_file=db_file), ) # Initialize storage storage = SqliteStorage(table_name="agent_sessions", db_file=db_file) # 定义工具函数 def get_current_time(): """获取当前时间的工具函数""" import datetime current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") return f"当前时间是: {current_time}" def remember_info(info: str): """主动记住信息的工具函数""" return f"我已经记住了这个信息: {info}" # Create Agent with Memory and Storage agent = Agent( model=model, # Store memories in a database memory=memory, # Give the Agent the ability to update memories enable_agentic_memory=True, # Run the MemoryManager after each response enable_user_memories=True, # Store the chat history in the database storage=storage, # Add the chat history to the messages add_history_to_messages=True, # Number of history runs to include num_history_runs=3, # Tools tools=[ # YFinanceTools(stock_price=True, company_info=True), find_contact_by_name, transfer_money, # get_current_time, # remember_info, get_account_balance ], markdown=False, # 终端模式关闭markdown show_tool_calls=True, instructions=""" 你是一个具有记忆功能的智能AI助手,检测到用户有转账意图的时候会收集转账所需的信息(转账需要收款人姓名、收款人电话、转账金额)如果获得了收款人的姓名,可以根据收款人的姓名调用{find_contact_by_name}去获取对应的收款人电话再调用{get_account_balance}获取账户余额,如果用户需要转账的余额大于账户余额,则提示用户账户余额不足,如果没有对应的收款人电话再询问用户,基于通义千问模型。你的特点: 💬 **对话原则**: - 回答简洁明了,避免冗长 - 你只使用markdown格式化输出 - 当你收集完成转账信息后,会用JSON格式输出,并询问用户是否确认转账 - 如果用户确认转账,则调用转账工具{transfer_money}函数进行转账,并输出转账结果 - 如果用户不确认转账,则不进行转账 让我们开始愉快的对话吧!我会记住我们的每次交流。 """, ) print("✅ Memory Agent 创建成功!") print(f"📱 模型: qwen3-32b") print(f"🧠 记忆: SQLite数据库 ({db_file})") print(f"💾 存储: 会话历史记录") print(f"👤 用户ID: {user_id}") return agent, user_id, memory except Exception as e: print(f"❌ 创建Agent失败: {e}") print("💡 可能的原因:") print(" - API连接问题") print(" - 模型配置错误") print(" - 依赖包未正确安装") return None def print_chat_banner(): """打印对话横幅""" banner = """ 🧠 具有Memory功能的智能AI助手 =============================== 💡 使用说明: - 直接输入问题开始对话 - 输入 'quit' 或 'exit' 退出 - 输入 'help' 查看更多帮助 - 输入 'clear' 清屏 - 输入 'memory' 查看记忆状态 🔥 功能特色: - 🧠 持久化记忆 - 跨会话记住您的信息 - 🤖 智能对话 - 通义千问强大推理 - 📊 实时数据 - 股票信息查询 - 🎯 个性化 - 基于记忆提供定制建议 - 💾 历史保存 - 自动保存对话历史 """ print(banner) def handle_special_commands(user_input: str) -> str: """处理特殊命令""" cmd = user_input.lower().strip() if cmd in ['quit', 'exit', 'bye', 'q']: return 'quit' elif cmd == 'help': help_text = """ 🆘 命令帮助 =========== 📝 基本命令: - quit/exit/bye/q : 退出对话 - help : 显示此帮助 - clear : 清屏 - memory : 查看记忆状态 """ print(help_text) return 'help' elif cmd == 'clear': os.system('clear' if os.name == 'posix' else 'cls') print_chat_banner() return 'clear' elif cmd == 'memory': return 'memory' return 'continue' def print_memory_status(memory, user_id): """显示记忆状态""" try: print("\n🧠 Memory状态:") print(f" 用户ID: {user_id}") print(f" 数据库: tmp/agent_memory.db") # 尝试获取记忆信息 try: memories = memory.get_user_memories(user_id=user_id) print(f" 记忆数量: {len(memories) if memories else 0}") if memories: print("\n📝 最近的记忆:") for i, mem in enumerate(memories[-3:], 1): # 显示最近3条记忆 content = mem.get('content', '')[:100] # 限制显示长度 print(f" {i}. {content}...") else: print(" 📭 暂无记忆内容") except Exception as e: print(f" 记忆系统: 已启用 (详情获取失败: {e})") print("\n💭 记忆功能可以帮助我:") print(" - 记住您的姓名、偏好和兴趣") print(" - 保持跨会话的对话连贯性") print(" - 提供基于历史的个性化建议") print(" - 回忆之前讨论过的话题") except Exception as e: print(f"❌ 无法获取记忆状态: {e}") def chat_with_memory_agent(agent, user_id, memory): """开始与Memory Agent对话""" if not agent: print("❌ Agent未创建成功,无法开始对话") return print_chat_banner() print("🎉 Memory Agent已就绪,开始对话吧!") conversation_count = 0 try: while True: # 获取用户输入 try: user_input = input(f"🙋 您 ({conversation_count+1}): ").strip() except (EOFError, KeyboardInterrupt): print("\n👋 对话结束,再见!") break # 跳过空输入 if not user_input: continue # 处理特殊命令 cmd_result = handle_special_commands(user_input) if cmd_result == 'quit': print("👋 感谢使用Memory Agent,我会记住我们的对话!") break elif cmd_result in ['help', 'clear']: continue elif cmd_result == 'memory': print_memory_status(memory, user_id) continue # Agent处理并回复 conversation_count += 1 print(f"\n🤖 AI助手 ({conversation_count}):") print("=" * 60) try: # 调用Agent处理用户问题,使用user_id来关联记忆 response = agent.print_response( user_input, user_id=user_id, # 关联用户ID stream=True ) print("=" * 60) print() # 空行分隔 # 每3轮对话提示一次记忆功能 if conversation_count % 3 == 0: print("🧠 提示:我已经记住了我们的对话内容,下次聊天时我仍会记得!") except KeyboardInterrupt: print("\n⚠️ 回复被中断,继续下一轮对话...") continue except Exception as e: print(f"❌ 处理消息出错: {e}") print("💡 请检查网络连接或API配置") continue except Exception as e: print(f"❌ 对话过程中出现错误: {e}") print("🔚 对话会话结束,记忆已保存") def main(): """主函数""" try: print("🚀 启动具有Memory功能的AI对话系统...") # 创建Memory Agent result = create_memory_agent() if not result: sys.exit(1) agent, user_id, memory = result print("\n" + "="*50) print("🚀 启动对话模式...") # 开始对话 chat_with_memory_agent(agent, user_id, memory) except Exception as e: print(f"❌ 启动失败: {e}") sys.exit(1) if __name__ == "__main__": main()