tools/web_search.py

# filename: web_search.py (Hybrid Intelligence Version)

import os
import asyncio
from tavily import TavilyClient
from anthropic import AsyncAnthropic
from dotenv import load_dotenv

load_dotenv()

tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
async_anthropic_client = AsyncAnthropic()

async def get_travel_intelligence_briefing(destination: str, time_period: str, user_keywords: list[str] = None) -> str:
    """
    接收目的地、时间段和用户特定的关键词，执行一个混合了
    “基础情报”和“关键词情报”的网络搜索，然后汇总成一份简报。
    """
    print(f"\n--- [Hybrid Intelligence] Briefing for: {destination} ({time_period}) with keywords: {user_keywords} ---")

    queries = [
        f"major holidays or festivals in {destination} during {time_period}",
        f"weather advisories or natural disaster risk for {destination} {time_period}",
        f"general transport strikes or protests planned for {destination} {time_period}",
        f"official travel warnings or geopolitical issues for {destination}",
    ]

    if user_keywords:
        for keyword in user_keywords:
            queries.append(f"latest news, closures, or strikes related to {keyword} {destination}")

    print(f"--- Step 1: Generated {len(queries)} hybrid queries. ---")

    async def _run_single_search(query):
        try:
            return await asyncio.to_thread(
                tavily_client.search,
                query=query, search_depth="basic", max_results=3
            )
        except Exception as e:
            print(f"  - Search failed for query '{query}': {e}")
            return None

    print("--- Step 2: Executing all searches in parallel...")
    search_tasks = [_run_single_search(q) for q in queries]
    all_results = await asyncio.gather(*search_tasks)

    raw_context = ""
    for query, result_pack in zip(queries, all_results):
        if result_pack and result_pack.get('results'):
            raw_context += f"Intelligence regarding '{query}':\n"
            for res in result_pack['results']:
                raw_context += f"- {res['content']}\n"
            raw_context += "\n---\n"

    if not raw_context:
        return "Could not retrieve any relevant travel intelligence from the web."

    print("--- Step 3: Synthesizing all intelligence into a final briefing...")
    system_prompt = "You are a professional travel intelligence analyst. Your job is to synthesize raw data into a clear, concise, and actionable briefing for a trip planner. Structure your output with clear headings."

    synthesis_prompt = f"""
    Please analyze the following raw web search results for a trip to **{destination}** during **{time_period}**.
    Consolidate all information into a structured 'Travel Intelligence Briefing'.
    For each category (General Safety, Weather, Events, Transportation, and **User-Specific Keywords**), provide a concise summary.
    If no information is found for a category, state "No significant information found."

    **Raw Intelligence Data:**
    ---
    {raw_context}
    ---

    **Required Output Format:**

    **Travel Intelligence Briefing: {destination} ({time_period})**

    **1. General Safety & Advisories:**
    - [Summary of wars, warnings, policy changes]

    **2. Weather & Environment:**
    - [Summary of forecasts, climate patterns, disaster risks]

    **3. Cultural Events & Holidays:**
    - [Summary of festivals, holidays, major events]

    **4. General Transportation:**
    - [Summary of general strikes or disruptions]

    **5. User-Specific Items:**
    - [For each keyword from the user_keywords list, provide a specific summary. e.g., "Regarding Finnair: ...", "Regarding Louvre Museum: ..."]
    """

    try:
        response = await async_anthropic_client.messages.create(
            model="claude-3-haiku-20240307", max_tokens=1000,
            system=system_prompt, messages=[{"role": "user", "content": synthesis_prompt}]
        )
        return response.content[0].text
    except Exception as e:
        return f"Error during final synthesis: {e}"