"""Инструменты веб-исследования для deep agent.""" from __future__ import annotations import os import httpx from langchain.tools import tool from markdownify import markdownify _BROWSER_HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" ), } @tool def get_page_content(url: str) -> str: """Загружает веб-страницу по URL и возвращает текст в markdown.""" try: with httpx.Client(timeout=30.0, follow_redirects=True, headers=_BROWSER_HEADERS) as client: response = client.get(url) if response.status_code != 200: return f"Ошибка {response.status_code} при загрузке {url}" return markdownify(response.text)[:12000] except Exception as exc: return f"Не удалось загрузить страницу: {exc}" @tool def web_search(query: str, max_results: int = 5) -> str: """Ищет информацию в интернете (Tavily или DuckDuckGo как запасной вариант). Args: query: поисковый запрос max_results: число результатов """ tavily_key = os.getenv("TAVILY_API_KEY", "").strip() if tavily_key: try: from tavily import TavilyClient client = TavilyClient(api_key=tavily_key) resp = client.search(query, max_results=max_results) parts = [] for item in resp.get("results", []): parts.append( f"**{item.get('title', '')}**\n" f"url: {item.get('url', '')}\n" f"content: {item.get('content', '')}\n" ) if parts: return "\n".join(parts) except Exception as exc: return f"Ошибка Tavily: {exc}" try: from duckduckgo_search import DDGS lines = [] with DDGS() as ddgs: for item in ddgs.text(query, max_results=max_results): lines.append( f"**{item.get('title', '')}**\n" f"url: {item.get('href', '')}\n" f"content: {item.get('body', '')}\n" ) return "\n".join(lines) if lines else "Ничего не найдено" except Exception as exc: return ( "Поиск недоступен. Задайте TAVILY_API_KEY в .env " f"или установите duckduckgo-search. Ошибка: {exc}" )