File size: 2,898 Bytes
5ab87e0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
from __future__ import annotations
import asyncio
from typing import Dict, List
from utils import (google_search, url_hits_to_markdown,
search_result_to_markdown,
async_search_and_extract, _bad)
from fetchers_async import fetch_url
from compressor import compress_text, query_text
from config import CFG
import logging
# print("PYTHONPATH:", sys.path)
def web_search(query):
return search_urls(query = query, top_k = 10)
def web_visit(url):
return open_url(url = url, compress = False)
# ── 1. search_urls ──────────────────────────────────────────────────────
def search_urls(query: str, top_k: int = 10) -> str:
return url_hits_to_markdown(google_search(query, top_k))
# ── 2. open_url ─────────────────────────────────────────────────────────
def open_url(url: str, *, compress: bool = True, pct: float = CFG.pct,
model: str = "gpt-4o-mini") -> str:
if _bad(url): return _bad(url)
try:
body = asyncio.run(fetch_url(url))
body = str(body)
except Exception as e:
return f"[error fetching URL: {e}]"
if compress:
try:
body = compress_text(body, pct=pct, model=model)
except Exception as e:
body = f"[compression failed: {e}]\n\n{body[:2000]}"
return body
# ── 3. search_and_parse_query ───────────────────────────────────────────
def search_and_parse_query(query: str, top_k: int = 3, *,
compress: bool = True, pct: float = CFG.pct) -> str:
blocks = asyncio.run(async_search_and_extract(query, top_k))
if compress:
for b in blocks:
try:
cmp = compress_text(b["body"], pct=pct)
b["body"] = (f"**Summary:**\n{cmp['narrative']}\n\n"
f"**Facts:**\n{cmp['facts']}\n\n"
f"**Tables:**\n{cmp['tables']}")
except Exception as e:
b["body"] = f"[compression failed: {e}]\n\n{b['body']}"
return search_result_to_markdown(blocks)
# ── 4. query_url ────────────────────────────────────────────────────────
def query_url(url: str, goal: str, *, model: str = "gpt-4.1-mini") -> str:
if _bad(url): return _bad(url)
body = asyncio.run(fetch_url(url))
if not body or body.startswith("[error"):
return f"[failed to retrieve content from {url}]\n\n{body}"
return query_text(url, body, goal, model=model)['extracted_info']
|