{ "cells": [ { "cell_type": "code", "execution_count": 34, "id": "938b2e36-f95c-4b0f-8770-335c6bb5bc0e", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "API 配置完成!\n" ] } ], "source": [ "import google.generativeai as genai\n", "import requests\n", "import json\n", "import os\n", "import re\n", "from tavily import TavilyClient\n", "from IPython.display import display, Markdown, HTML, clear_output\n", "import time\n", "\n", "# 配置 API 密钥\n", "GEMINI_API_KEY = \"AIzaSyBYXm6uWt5-wDi9wXQ1oBhHhsMTWAH8kbo\"\n", "TAVILY_API_KEY = \"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\"\n", "MODEL_NAME = \"gemini-2.5-flash\" # 如果2.5不可用,使用1.5\n", "\n", "# 配置 Gemini\n", "genai.configure(api_key=GEMINI_API_KEY)\n", "\n", "# 配置 Tavily\n", "os.environ['TAVILY_API_KEY'] = TAVILY_API_KEY\n", "\n", "print(\"API 配置完成!\")" ] }, { "cell_type": "code", "execution_count": 35, "id": "66d9d404-3c95-42f2-8975-436769b3cb87", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "✅ 工具函数定义完成!\n" ] } ], "source": [ "# 单元块 2: 定义工具函数\n", "def get_weather(city: str) -> str:\n", " \"\"\"查询城市天气\"\"\"\n", " url = f\"https://wttr.in/{city}?format=j1\"\n", " \n", " try:\n", " response = requests.get(url)\n", " response.raise_for_status()\n", " data = response.json()\n", " \n", " current_condition = data['current_condition'][0]\n", " weather_desc = current_condition['weatherDesc'][0]['value']\n", " temp_c = current_condition['temp_C']\n", " humidity = current_condition['humidity']\n", " wind_speed = current_condition['windspeedKmph']\n", " \n", " return f\"{city}当前天气:{weather_desc},气温{temp_c}°C,湿度{humidity}%,风速{wind_speed}km/h\"\n", " \n", " except Exception as e:\n", " return f\"错误:查询天气时遇到问题 - {e}\"\n", "\n", "def get_attraction(city: str, weather: str) -> str:\n", " \"\"\"查询景点推荐\"\"\"\n", " try:\n", " tavily = TavilyClient(api_key=os.environ['TAVILY_API_KEY'])\n", " query = f\"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由\"\n", " \n", " response = tavily.search(query=query, search_depth=\"basic\", include_answer=True)\n", " \n", " if response.get(\"answer\"):\n", " return response[\"answer\"]\n", " \n", " formatted_results = []\n", " for result in response.get(\"results\", []):\n", " formatted_results.append(f\"- {result['title']}: {result['content']}\")\n", " \n", " if not formatted_results:\n", " return \"抱歉,没有找到相关的旅游景点推荐。\"\n", "\n", " return \"根据搜索,为您找到以下信息:\\n\" + \"\\n\".join(formatted_results)\n", "\n", " except Exception as e:\n", " return f\"错误:执行搜索时出现问题 - {e}\"\n", "\n", "print(\"✅ 工具函数定义完成!\")" ] }, { "cell_type": "code", "execution_count": 36, "id": "e953fee4-9e3c-4e34-bf48-4ea002c3bb92", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "✅ 智能助手类定义完成!\n" ] } ], "source": [ "# 单元块 3: 定义智能助手类\n", "class NotebookTravelAssistant:\n", " def __init__(self, model_name=\"gemini-1.5-flash\"):\n", " # 初始化模型\n", " self.model_name = model_name\n", " try:\n", " self.model = genai.GenerativeModel(model_name)\n", " except:\n", " print(f\"⚠️ 模型 {model_name} 不可用,使用 gemini-1.5-flash 替代\")\n", " self.model = genai.GenerativeModel(\"gemini-1.5-flash\")\n", " \n", " # 定义工具\n", " self.available_tools = {\n", " \"get_weather\": get_weather,\n", " \"get_attraction\": get_attraction,\n", " }\n", " \n", " # 系统提示词\n", " self.system_prompt = \"\"\"\n", "你是一个智能旅行助手。你的任务是分析用户的请求,并使用可用工具一步步地解决问题。\n", "\n", "# 可用工具:\n", "- `get_weather(city: str)`: 查询指定城市的实时天气。\n", "- `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。\n", "\n", "# 行动格式:\n", "你的回答必须严格遵循以下格式。首先是你的思考过程,然后是你要执行的具体行动。\n", "Thought: [这里是你的思考过程和下一步计划]\n", "Action: [这里是你要调用的工具,格式为 function_name(arg_name=\"arg_value\")]\n", "\n", "# 任务完成:\n", "当你收集到足够的信息,能够回答用户的最终问题时,你必须使用 `finish(answer=\"...\")` 来输出最终答案。\n", "\n", "请开始吧!\n", "\"\"\"\n", " \n", " def generate_response(self, prompt: str) -> str:\n", " \"\"\"调用Gemini生成回应\"\"\"\n", " full_prompt = f\"{self.system_prompt}\\n\\n用户请求: {prompt}\"\n", " response = self.model.generate_content(full_prompt)\n", " return response.text\n", " \n", " def parse_action(self, text: str):\n", " \"\"\"解析行动指令\"\"\"\n", " action_match = re.search(r\"Action: (.*)\", text, re.DOTALL)\n", " if not action_match:\n", " return None, None\n", " \n", " action_str = action_match.group(1).strip()\n", " \n", " if action_str.startswith(\"finish\"):\n", " final_answer_match = re.search(r'finish\\(answer=\"(.*)\"\\)', action_str)\n", " if final_answer_match:\n", " return \"finish\", final_answer_match.group(1)\n", " return \"finish\", \"抱歉,我无法提供完整的答案。\"\n", " \n", " tool_match = re.search(r\"(\\w+)\\((.*)\\)\", action_str)\n", " if not tool_match:\n", " return None, None\n", " \n", " tool_name = tool_match.group(1)\n", " args_str = tool_match.group(2)\n", " \n", " kwargs = {}\n", " if args_str:\n", " arg_matches = re.findall(r'(\\w+)=\"([^\"]*)\"', args_str)\n", " kwargs = dict(arg_matches)\n", " \n", " return tool_name, kwargs\n", "\n", "print(\"✅ 智能助手类定义完成!\")" ] }, { "cell_type": "code", "execution_count": 37, "id": "ab5d3142-c119-46ad-a7a1-ec1aa2e79435", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "✅ 显示函数定义完成!\n" ] } ], "source": [ "# 单元块 4: 定义显示函数\n", "def display_user_input(user_input):\n", " \"\"\"显示用户输入\"\"\"\n", " display(HTML(f\"\"\"\n", "