给 CrewAI 的 Crew 接上实时网页搜索

import os, httpx
from crewai.tools import tool

API_KEY = os.environ["APIPICK_KEY"]

@tool("Web Search")
def web_search(query: str) -> str:
    """搜索实时网络并返回排好序的结果。

    当答案依赖当前信息时调用它——新闻、价格、产品细节，
    以及任何在你训练数据之后才发布的东西。查询词用大白话，
    按普通人往搜索框里敲的方式写。

    返回最多五条，格式为 'title — url — snippet' 的行。
    """
    r = httpx.post(
        "https://www.apipick.com/api/search/web",
        headers={"x-api-key": API_KEY},
        json={"query": query, "max_num_results": 5},
        timeout=20,
    )
    r.raise_for_status()
    return "\n\n".join(
        f"{x['title']} — {x['url']}\n{x['snippet']}"
        for x in r.json()["results"]
    )

from typing import Type, Optional
from pydantic import BaseModel, Field
from crewai.tools import BaseTool

class WebSearchInput(BaseModel):
    query: str = Field(..., description="大白话的搜索词")
    country_code: Optional[str] = Field(
        None, description="用于本地化结果的 ISO 国家码，例如 US 或 GB"
    )
    start_date: Optional[str] = Field(
        None, description="只要这个 ISO 日期（YYYY-MM-DD）及之后发布的结果"
    )

class WebSearchTool(BaseTool):
    name: str = "Web Search"
    description: str = (
        "搜索实时网络获取当前信息。用于新闻、价格，以及任何在训练截止"
        "之后发布的内容。可选按国家或发布日期限定。"
    )
    args_schema: Type[BaseModel] = WebSearchInput

    def _run(self, query: str, country_code=None, start_date=None) -> str:
        body = {"query": query, "max_num_results": 5}
        if country_code: body["country_code"] = country_code
        if start_date:   body["start_date"] = start_date
        r = httpx.post(
            "https://www.apipick.com/api/search/web",
            headers={"x-api-key": API_KEY}, json=body, timeout=20,
        )
        r.raise_for_status()
        return "\n\n".join(
            f"{x['title']} — {x['url']}\n{x['snippet']}"
            for x in r.json()["results"]
        )

from crewai import Agent, Task, Crew

search = WebSearchTool()

researcher = Agent(
    role="Researcher",
    goal="为当前问题找到并引用当前的证据",
    backstory="没有 URL 支撑的事实，你从不陈述。",
    tools=[search],          # 只有这个 Agent 会搜索
)

analyst = Agent(
    role="Analyst",
    goal="把 researcher 的证据变成一个站得住的结论",
    backstory="你严格只基于交到你手上的证据工作。",
    tools=[],                # 不给工具——读 researcher 的产出
)

crew = Crew(
    agents=[researcher, analyst],
    tasks=[
        Task(description="研究：{topic}", agent=researcher,
             expected_output="5–8 条要点，每条附 URL"),
        Task(description="仅基于上述发现给出结论", agent=analyst,
             expected_output="一份带行内引用的简短备忘"),
    ],
)

crew.kickoff(inputs={"topic": "今年欧盟 AI 法案时间表的变化"})

@tool("Read Pages")
def read_pages(urls: list[str]) -> str:
    """取回指定 URL 的干净正文。

    在 Web Search 之后、摘要不足以回答时调用。只传真正相关的
    URL——最多三个。
    """
    r = httpx.post(
        "https://www.apipick.com/api/extract",
        headers={"x-api-key": API_KEY},
        json={"urls": urls[:3]},
        timeout=60,
    )
    r.raise_for_status()
    return "\n\n---\n\n".join(
        f"# {x['title']}\n{x['url']}\n\n{x['content'][:6000]}"
        for x in r.json()["results"] if x["status"] == "ok"
    )