Агент-исследователь для веб-поиска

Цель проекта

Необходимые инструменты

Step 1: Search Tool

import requests

def search(query: str, k: int = 5) -> list[dict]:
    r = requests.post(
        'https://api.tavily.com/search',
        json={'api_key': TAVILY_KEY, 'query': query, 'max_results': k}
    )
    return r.json()['results']   # list of {url, title, snippet}

Step 2: Fetch Tool

from bs4 import BeautifulSoup

def fetch_url(url: str) -> str:
    r = requests.get(url, timeout=10)
    soup = BeautifulSoup(r.text, 'html.parser')
    for tag in soup(['script', 'style', 'nav', 'footer']):
        tag.decompose()
    return soup.get_text(separator='\n', strip=True)[:5000]   # truncate

Step 3: Tool Definitions

tools = [
    {'type': 'function', 'function': {
        'name': 'search',
        'description': 'Search the web for a query, return top results with URLs',
        'parameters': {'type': 'object', 'properties': {'query': {'type': 'string'}}, 'required': ['query']}
    }},
    {'type': 'function', 'function': {
        'name': 'fetch_url',
        'description': 'Download the text of a URL',
        'parameters': {'type': 'object', 'properties': {'url': {'type': 'string'}}, 'required': ['url']}
    }}
]
import json
print(json.dumps(tools, indent=2))

Step 4: The Agent Loop

def research(question, max_steps=8):
    messages = [
        {'role': 'system', 'content': 'You are a research agent. Use search and fetch_url to gather facts. Cite each claim with a URL.'},
        {'role': 'user', 'content': question}
    ]
    for _ in range(max_steps):
        r = oai.chat.completions.create(model='gpt-4o-mini', messages=messages, tools=tools)
        msg = r.choices[0].message
        messages.append(msg)
        if not msg.tool_calls:
            return msg.content
        for tc in msg.tool_calls:
            result = dispatch(tc)
            messages.append({'role': 'tool', 'tool_call_id': tc.id, 'content': json.dumps(result)})
    return 'Step limit reached'

Tool Dispatcher

def dispatch(tc):
    args = json.loads(tc.function.arguments)
    if tc.function.name == 'search':
        return search(**args)
    if tc.function.name == 'fetch_url':
        return fetch_url(**args)
    return {'error': 'unknown tool'}

Ограничьте число шагов

Обрезайте загруженные страницы

Учитывайте robots.txt

Кэшируйте результаты

@lru_cache(maxsize=1000)
def cached_fetch(url):
    return fetch_url(url)

Обрабатывайте сбои

try:
    return fetch_url(url)
except Exception as e:
    return {'error': f'Could not fetch {url}: {e}'}

Принудительно добавляйте цитаты

Следите за стоимостью

Ограничение числа шагов

Повторим