Files
task-69de7223f309a98be0007e09/research_tools.py
T
2026-05-26 07:51:28 +00:00

75 lines
2.6 KiB
Python

"""Инструменты веб-исследования для 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}"
)