Переглянути джерело

agno框架转账小玩具

linzhaoxin319319 2 тижнів тому
батько
коміт
e1df86603e
1 змінених файлів з 328 додано та 0 видалено
  1. 328 0
      林兆新/2/chat_with_agent_intructions.py

+ 328 - 0
林兆新/2/chat_with_agent_intructions.py

@@ -0,0 +1,328 @@
+#!/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()