#!/usr/bin/env python3 """ 交互式Agent对话脚本 基于agno框架创建的终端对话界面 """ from agno.agent import Agent, RunResponse, RunResponseEvent from agno.models.openai import OpenAILike from agno.utils.pprint import pprint_run_response import os import sys import json import re from typing import Iterator import dotenv # 加载环境变量 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"} ] from typing import Optional def find_contact_by_name(name: str) -> Optional[str]: """根据姓名查找联系人电话号码""" for contact in CONTACTS: if contact["name"] == name: return contact["phone"] return None def create_agent(): """创建并配置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}}, ) def get_news(description: str): """示例工具函数 - 获取新闻""" print(f"[工具调用] get_news: {description}") return "暂无相关新闻" def extract_transfer_info(user_message: str) -> str: """从用户消息中提取转账信息 Args: user_message: 用户的自然语言输入 Returns: JSON字符串,包含提取的信息: { "is_transfer": true/false, "recipient_name": "姓名" (如果找到), "amount": "金额" (如果找到), "confidence": 0.0-1.0 (置信度) } """ return f"请分析以下用户消息,判断是否包含转账意图,并提取相关信息:'{user_message}'" agent = Agent( model=model, tool_choice="auto", tools=[get_news, extract_transfer_info], add_datetime_to_instructions=True, show_tool_calls=True, markdown=True, instructions=""" 你是一个智能助手,能够理解用户的各种需求。当用户表达转账意图时,你需要: 1. 识别转账关键词(如:转账、转钱、给...钱、汇款等) 2. 提取收款人姓名 3. 提取转账金额(如果有的话) 4. 返回JSON格式的结果 对于转账相关的消息,请使用extract_transfer_info工具来分析。 JSON格式示例: { "is_transfer": true, "recipient_name": "张三", "amount": "100", "confidence": 0.9 } 如果不是转账相关的消息,正常回答用户问题。 """, ) return agent def check_transfer_intent(user_input: str, agent: Agent) -> tuple[bool, Optional[str], Optional[str]]: """使用Agent检查用户输入是否包含转账意图 Returns: tuple: (是否转账, 收款人姓名, 金额) """ try: # 让Agent分析用户输入 response = agent.run(f"请分析这句话是否包含转账意图,并提取相关信息:'{user_input}'") # 从Agent响应中提取信息 response_text = "" if hasattr(response, 'content') and response.content: response_text = str(response.content) elif hasattr(response, 'messages') and response.messages: for msg in response.messages: if hasattr(msg, 'content') and msg.content: response_text += str(msg.content) # 尝试从响应中解析JSON import json # 查找JSON格式的数据 json_match = re.search(r'\{[^}]*"is_transfer"[^}]*\}', response_text) if json_match: transfer_info = json.loads(json_match.group()) is_transfer = transfer_info.get("is_transfer", False) recipient_name = transfer_info.get("recipient_name") amount = transfer_info.get("amount") confidence = transfer_info.get("confidence", 0.0) print(f"🔍 AI分析结果: 转账意图={is_transfer}, 置信度={confidence}") if recipient_name: print(f"📝 提取信息: 收款人={recipient_name}, 金额={amount or '未指定'}") # 只有在高置信度且确实是转账意图时才返回True if is_transfer and confidence > 0.7: return True, recipient_name, amount return False, None, None except Exception as e: print(f"⚠️ AI分析过程中出错: {e}") # 如果AI分析失败,回退到关键词检测 transfer_keywords = ["转账", "转钱", "汇款", "给钱", "付款", "支付"] if any(keyword in user_input for keyword in transfer_keywords): print("🔍 使用关键词检测到转账意图") return True, None, None return False, None, None def get_missing_info_via_agent(agent: Agent, prompt: str) -> Optional[str]: """使用Agent获取缺失的信息 Args: agent: Agent实例 prompt: 询问提示 Returns: 用户提供的信息,如果无法获取则返回None """ try: print(f"\n🤖 Agent: {prompt}") user_response = input("🙋 您: ").strip() if not user_response: return None # 使用Agent分析用户响应 analysis_prompt = f"用户说:'{user_response}',请提取出用户想要表达的具体信息(如姓名、金额、电话号码等),只返回提取到的关键信息,不要额外解释。" response = agent.run(analysis_prompt) # 从Agent响应中提取信息 response_text = "" if hasattr(response, 'content') and response.content: response_text = str(response.content) elif hasattr(response, 'messages') and response.messages: for msg in response.messages: if hasattr(msg, 'content') and msg.content: response_text += str(msg.content) # 简单清理响应文本,提取关键信息 cleaned_response = response_text.strip() # 移除常见的解释性词语 for phrase in ["用户想要表达的是", "提取到的信息是", "关键信息是", "具体信息是"]: cleaned_response = cleaned_response.replace(phrase, "").strip() return cleaned_response if cleaned_response else user_response except Exception as e: print(f"⚠️ AI分析用户输入时出错: {e}") return user_response if 'user_response' in locals() else None def handle_transfer_with_extracted_info(agent: Agent, name: Optional[str] = None, amount: Optional[str] = None): """处理转账功能(使用已提取的信息和Agent补全缺失信息)""" # 如果没有提取到姓名,使用Agent询问 if not name: print("📝 需要获取收款人信息...") name = get_missing_info_via_agent(agent, "请告诉我您要转账给谁?(请提供收款人的姓名)") if not name: print("⚠️ 无法获取收款人姓名,转账操作取消") return print(f"\n💰 准备转账给: {name}") # 查找联系人电话号码 phone = find_contact_by_name(name) if phone: print(f"📞 找到联系人电话: {phone}") else: print(f"❌ 联系人列表中未找到 {name} 的电话号码") phone = get_missing_info_via_agent(agent, f"请提供 {name} 的电话号码:") if not phone: print("⚠️ 无法获取电话号码,转账操作取消") return # 验证电话号码格式 # 先去除所有非数字字符,只保留数字 def normalize_phone(phone_str): return re.sub(r'\D', '', phone_str or "") while phone: normalized_phone = normalize_phone(phone) if re.match(r"^1\d{10}$", normalized_phone): phone = normalized_phone # 用标准化后的号码 break print("⚠️ 电话号码格式不正确,您之前输入的电话号码是:", phone) phone = get_missing_info_via_agent(agent, "请提供正确格式的电话号码(如:138-0000-1001 或 13800001001):") if not phone: print("⚠️ 无法获取有效电话号码,转账操作取消") return # 处理转账金额 validated_amount = None if amount: # 验证AI提取的金额 try: amount_float = float(amount) if amount_float > 0: validated_amount = amount print(f"💵 转账金额: ¥{amount}") else: print(f"⚠️ 提取的金额 {amount} 无效") amount = None # 重置amount,触发重新获取 except ValueError: print(f"⚠️ 提取的金额 {amount} 格式错误") amount = None # 重置amount,触发重新获取 # 如果没有有效金额,使用Agent询问 if not validated_amount: print("💵 需要获取转账金额...") while True: amount_str = get_missing_info_via_agent(agent, f"请告诉我您要转账多少钱给 {name}?(请输入数字金额)") if not amount_str: print("⚠️ 无法获取转账金额,转账操作取消") return try: # 从用户输入中提取数字 number_match = re.search(r'\d+(?:\.\d+)?', amount_str) if number_match: amount_float = float(number_match.group()) if amount_float > 0: validated_amount = str(amount_float) print(f"✅ 确认转账金额: ¥{validated_amount}") break else: print("⚠️ 转账金额必须大于0") else: print("⚠️ 无法从输入中识别有效的金额数字") except ValueError: print("⚠️ 金额格式错误,请重新输入") # 创建转账信息JSON transfer_info = { "recipient_name": name, "recipient_phone": phone, "amount": validated_amount, "timestamp": __import__('datetime').datetime.now().isoformat() } # 显示转账信息 print("\n" + "="*50) print("📋 转账信息确认:") print(f" 收款人: {transfer_info['recipient_name']}") print(f" 电话号码: {transfer_info['recipient_phone']}") print(f" 转账金额: ¥{transfer_info['amount']}") print(f" 时间: {transfer_info['timestamp']}") print("="*50) # 确认转账 confirm_response = input("请确认以上转账信息是否正确?(回答:y/n)").strip() if confirm_response: confirm_response_str_raw = str(confirm_response).strip().lower() if confirm_response_str_raw == "y": # 用户确认,保存转账信息 try: transfers_file = "transfer_records.json" if os.path.exists(transfers_file): with open(transfers_file, 'r', encoding='utf-8') as f: transfers = json.load(f) else: transfers = [] # 添加新转账记录,并附加用户确认回复 transfer_info_with_confirm = dict(transfer_info) transfer_info_with_confirm["user_confirm_response"] = '{"is_transfer": true}' transfers.append(transfer_info_with_confirm) with open(transfers_file, 'w', encoding='utf-8') as f: json.dump(transfers, f, ensure_ascii=False, indent=2) print("✅ 转账信息已保存!") print(f"📄 转账记录已保存到: {transfers_file}") except Exception as e: print(f"❌ 保存转账记录时出错: {e}") elif confirm_response_str_raw == "n": print("❌ 用户未确认转账,未保存转账信息。") else: print("⚠️ 无法识别的回复,转账操作取消") else: print("⚠️ 未收到确认回复,转账操作取消") def print_welcome(): """打印欢迎信息""" print("=" * 60) print("🤖 欢迎使用Agent对话系统!") print("=" * 60) print("💡 使用说明:") print(" - 直接输入您的问题进行对话") print(" - 输入 'exit', 'quit', 'bye' 或 '退出' 结束对话") print(" - 输入 'clear' 或 '清屏' 清空屏幕") print(" - 输入 'help' 或 '帮助' 查看帮助信息") print(" - 说出转账相关的话(如:给张三转100块钱)进行转账操作") print("=" * 60) def print_help(): """打印帮助信息""" print("\n📋 帮助信息:") print(" exit/quit/bye/退出 - 退出程序") print(" clear/清屏 - 清空屏幕") print(" help/帮助 - 显示此帮助信息") print(" 转账相关自然语言 - AI智能识别转账意图") print(" 例:给张三转100块 - 转账功能示例") print(" 其他任何内容 - 与Agent对话") def clear_screen(): """清空屏幕""" os.system('cls' if os.name == 'nt' else 'clear') def chat_loop(): """主对话循环""" agent = create_agent() print_welcome() while True: try: # 获取用户输入 user_input = input("\n🙋 您: ").strip() # 检查退出命令 if user_input.lower() in ['exit', 'quit', 'bye', '退出']: print("\n👋 再见! 感谢使用Agent对话系统!") break # 检查清屏命令 if user_input.lower() in ['clear', '清屏']: clear_screen() print_welcome() continue # 检查帮助命令 if user_input.lower() in ['help', '帮助']: print_help() continue # 检查是否包含转账意图(让AI判断) is_transfer, name, amount = check_transfer_intent(user_input, agent) if is_transfer: handle_transfer_with_extracted_info(agent, name, amount) continue # 检查空输入 if not user_input: print("⚠️ 请输入您的问题或命令") continue print(f"\n🤖 Agent: ") print("-" * 50) # 运行agent并获取响应流 response_stream: Iterator[RunResponseEvent] = agent.run( user_input, stream=True ) # 打印响应 pprint_run_response(response_stream, markdown=True) except KeyboardInterrupt: print("\n\n👋 检测到 Ctrl+C,正在退出...") break except Exception as e: print(f"\n❌ 发生错误: {e}") print("请重试或联系技术支持") def main(): """主函数""" try: # 检查环境变量 if not os.getenv("BAILIAN_API_KEY") or not os.getenv("BAILIAN_API_BASE_URL"): print("❌ 错误: 请确保设置了以下环境变量:") print(" - BAILIAN_API_KEY") print(" - BAILIAN_API_BASE_URL") print("\n💡 您可以创建 .env 文件来设置这些变量") sys.exit(1) # 开始对话 chat_loop() except Exception as e: print(f"❌ 启动失败: {e}") sys.exit(1) if __name__ == "__main__": main()