Google 搜索框的"信息 Agent"时代：从链接列表到意图追踪

"""
mini_info_agent.py — 简易信息 Agent 原型
依赖：pip install requests schedule
运行：python mini_info_agent.py
"""

import requests
import schedule
import time
import json
from datetime import datetime

# ---------- 1. 意图解析层 ----------
def parse_intent(raw_query: str) -> dict:
    """
    把原始查询拆成结构化意图。
    实际生产中应调用 LLM，这里用规则演示逻辑。
    """
    intent = {
        "raw": raw_query,
        "topic": None,
        "action": "summarize",  # 默认动作：汇总
        "track": False,         # 是否需要持续追踪
    }

    # 简单规则：包含"变化/更新/最新"则开启追踪
    track_keywords = ["变化", "更新", "最新", "动态", "趋势"]
    if any(kw in raw_query for kw in track_keywords):
        intent["track"] = True
        intent["action"] = "track_and_summarize"

    # 提取主题（去掉追踪关键词后的核心词）
    topic = raw_query
    for kw in track_keywords:
        topic = topic.replace(kw, "")
    intent["topic"] = topic.strip()

    return intent


# ---------- 2. 执行层：多源查询 ----------
def search_web(topic: str, num_results: int = 5) -> list[dict]:
    """
    用 SerpAPI（或任何搜索 API）获取结果。
    这里用 Google Custom Search JSON API 演示。
    你需要替换为自己的 API_KEY 和 CX。
    """
    API_KEY = "YOUR_GOOGLE_API_KEY"   # ← 替换为真实 key
    CX = "YOUR_CUSTOM_SEARCH_CX"      # ← 替换为真实 cx

    url = "https://www.googleapis.com/customsearch/v1"
    params = {
        "key": API_KEY,
        "cx": CX,
        "q": topic,
        "num": num_results,
    }

    try:
        resp = requests.get(url, params=params, timeout=10)
        resp.raise_for_status()
        items = resp.json().get("items", [])
        return [{"title": i["title"], "snippet": i["snippet"], "link": i["link"]} for i in items]
    except Exception as e:
        print(f"[搜索失败] {e}")
        return []


def summarize_results(results: list[dict], intent: dict) -> str:
    """
    把多条搜索结果整合为一段摘要。
    生产环境应调用 LLM 做真正的摘要生成。
    """
    if not results:
        return f"未找到关于「{intent['topic']}」的相关信息。"

    lines = [f"📊 关于「{intent['topic']}」的整合摘要："]
    for i, r in enumerate(results, 1):
        lines.append(f"  {i}. {r['title']}")
        lines.append(f"     {r['snippet']}")
    lines.append(f"\n共整合 {len(results)} 条来源，查询时间 {datetime.now().strftime('%H:%M')}。")
    return "\n".join(lines)


# ---------- 3. 追踪层：定时检查变化 ----------
last_snapshot = []

def track_changes(topic: str):
    """对比新旧搜索结果，检测是否有变化"""
    global last_snapshot
    current = search_web(topic, num_results=3)

    # 简单变化检测：对比第一条结果的标题和摘要
    if last_snapshot and current:
        old_top = last_snapshot[0]["title"] + last_snapshot[0]["snippet"]
        new_top = current[0]["title"] + current[0]["snippet"]
        if old_top != new_top:
            print(f"\n🔔 [追踪通知] 「{topic}」有新变化！")
            print(f"  旧: {last_snapshot[0]['title']}")
            print(f"  新: {current[0]['title']}")
            print(f"  链接: {current[0]['link']}")
        else:
            print(f"  [{datetime.now().strftime('%H:%M')}] 无变化，继续监控…")
    else:
        print(f"  [{datetime.now().strftime('%H:%M')}] 首次快照已保存。")

    last_snapshot = current


# ---------- 主流程 ----------
def run_agent(query: str):
    intent = parse_intent(query)
    print(f"🧠 解析意图: {json.dumps(intent, ensure_ascii=False)}")

    results = search_web(intent["topic"])
    summary = summarize_results(results, intent)
    print(summary)

    if intent["track"]:
        print(f"\n⏳ 开启持续追踪，每 30 分钟检查一次「{intent['topic']}」…")
        # 先保存初始快照
        global last_snapshot
        last_snapshot = results
        schedule.every(30).minutes.do(track_changes, topic=intent["topic"])

        # 保持运行（演示用，生产环境应异步化）
        while True:
            schedule.run_pending()
            time.sleep(1)


# ---------- 入口 ----------
if __name__ == "__main__":
    # 示例查询：包含追踪关键词，Agent 会自动开启持续监控
    run_agent("日本签证政策最新变化")