当搜索从人转向机器：AnySearch 如何为 AI Agent 重造信息获取层

import os
import json
from openai import OpenAI

# --- 假设 AnySearch 提供 REST API，以下为推演的调用方式 ---
# 实际 endpoint、参数名和返回结构请以官方文档为准

ANYSEARCH_API_URL = "https://api.anysearch.ai/v1/search"
ANYSEARCH_API_KEY = os.environ.get("ANYSEARCH_API_KEY", "your-key-here")

def call_anysearch(query: str, mode: str = "fact", depth: str = "standard") -> dict:
    """
    调用 AnySearch 搜索接口。

    参数说明（基于发布信息推演）：
    - query: 搜索意图，支持自然语言但建议结构化
    - mode: 返回模式 — "fact" 返回事实/实体, "overview" 返回综合摘要
    - depth: 搜索深度 — "standard" 平衡速度与覆盖, "deep" 更全面但更慢
    """
    import requests

    headers = {
        "Authorization": f"Bearer {ANYSEARCH_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "query": query,
        "mode": mode,
        "depth": depth,
        # 结构化输出：要求返回可直接用于推理的 JSON
        "output_format": "structured",
    }

    resp = requests.post(ANYSEARCH_API_URL, headers=headers, json=payload, timeout=30)
    resp.raise_for_status()
    return resp.json()


def run_research_agent(topic: str) -> str:
    """
    一个简单的调研 Agent：搜索 → 整理 → 生成报告。
    对比传统做法：Agent 拿到 HTML 碎片后要自己清洗，
    这里直接拿到结构化数据，省掉后处理步骤。
    """
    # 第一步：用 AnySearch 获取结构化信息
    search_result = call_anysearch(
        query=f"{topic} 最新进展和关键数据",
        mode="fact",
        depth="standard",
    )

    # 假设返回结构如下（实际结构以官方文档为准）：
    # {
    #   "facts": [
    #     {"statement": "...", "source": "...", "confidence": 0.92},
    #     ...
    #   ],
    #   "entities": [
    #     {"name": "...", "type": "...", "attributes": {...}},
    #     ...
    #   ],
    #   "summary": "..."
    # }

    # 第二步：把结构化搜索结果直接注入 LLM prompt
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

    search_context = json.dumps(search_result, ensure_ascii=False, indent=2)

    prompt = f"""你是一位技术调研分析师。以下是关于「{topic}」的结构化搜索结果，
    数据已经过预处理，来源可信度已标注。请基于这些数据撰写一份简明调研摘要，
    包含：关键事实、数据指标、主要参与方、趋势判断。

    搜索结果：
    {search_context}

    要求：只使用搜索结果中的信息，不自行编造数据。如果某条事实 confidence < 0.7，
    在摘要中标注"待验证"。"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
    )

    return response.choices[0].message.content


# --- 运行示例 ---
if __name__ == "__main__":
    report = run_research_agent("Rust 在嵌入式系统中的应用")
    print(report)

# 传统搜索 API 的 Agent 后处理（现在不需要了）
def legacy_postprocess(raw_results: list[dict]) -> list[dict]:
    """传统搜索返回后，Agent 需要自己做的事"""
    cleaned = []
    for item in raw_results:
        # 1. 去广告和 SEO 垃圾 — 需要自建过滤规则
        if is_seo_spam(item["snippet"]):
            continue
        # 2. 提取正文 — 需要额外请求 + HTML 解析
        full_text = fetch_and_parse(item["url"])  # 慢、不稳定
        # 3. 判断可信度 — 需要自建评分逻辑
        score = heuristic_credibility(item["domain"], full_text)
        cleaned.append({"text": full_text, "credibility": score})
    return cleaned

def multi_turn_research(initial_topic: str, max_rounds: int = 3) -> str:
    """
    多轮调研 Agent：根据推理结果动态补充搜索。
    """
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
    accumulated_facts = []

    current_query = initial_topic

    for round_num in range(max_rounds):
        # 搜索：根据当前信息缺口调整 query 和 mode
        if round_num == 0:
            mode, depth = "overview", "standard"
        else:
            mode, depth = "fact", "deep"  # 后续轮次更聚焦、更深入

        result = call_anysearch(query=current_query, mode=mode, depth=depth)
        accumulated_facts.append(result)

        # 推理：判断信息是否足够，生成下一步搜索方向
        all_context = json.dumps(accumulated_facts, ensure_ascii=False, indent=2)

        reasoning_prompt = f"""基于已收集的信息，判断是否需要补充搜索。

        已收集信息（第 {round_num + 1} 轮）：
        {all_context}

        原始课题：{initial_topic}

        请回答：
        1. 当前信息是否足以完成调研？如果足够，直接输出最终报告。
        2. 如果不够，指出具体信息缺口，并给出下一轮搜索的精确 query。

        输出格式：
        {"status": "sufficient" | "need_more", "next_query": "...", "report": "..."}"""

        resp = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": reasoning_prompt}],
            temperature=0.2,
        )

        decision = json.loads(resp.choices[0].message.content)

        if decision["status"] == "sufficient":
            return decision["report"]

        current_query = decision["next_query"]

    # 达到最大轮次，用已有信息生成报告
    final_prompt = f"基于以下信息撰写调研报告：\n{all_context}"
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": final_prompt}],
        temperature=0.3,
    )
    return resp.choices[0].message.content