123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434 |
- #!/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()
|