WIP: feat: Most efficient deep research system ever made #32

Draft
typosaurus wants to merge 13 commits from typosaurus/31-most-efficient-deep-research-system-ever-made into main
pull from: typosaurus/31-most-efficient-deep-research-system-ever-made
merge into: typosaurus:main
Conversation 0 Commits 13 Files Changed 17 +259
4 changed files with 259 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 14a60ef77f - Show all commits

6
src/typosaurus_sandbox/research/__init__.py
View File

@ -3,6 +3,7 @@
from typosaurus_sandbox.research.cache import TTLCache
from typosaurus_sandbox.research.client import RsearchClient, RsearchError
from typosaurus_sandbox.research.config import ResearchConfig
from typosaurus_sandbox.research.engine import ResearchEngine, ResearchReport, RoundSummary
from typosaurus_sandbox.research.envelopes import (
ChatResponse,
ChatUsage,
@ -40,7 +41,10 @@ __all__ = [
"Extraction",
"PipelineReport",
"QueryFrontier",
"ResearchEngine",
"ResearchPipeline",
"ResearchReport",
"RoundSummary",
"RsearchClient",
"RsearchError",
"ResearchConfig",
@ -60,3 +64,5 @@ __all__ = [

42
src/typosaurus_sandbox/research/__main__.py Normal file
View File

@ -0,0 +1,42 @@
# retoor <retoor@molodetz.nl>
import argparse
import asyncio
import json
import logging
from typosaurus_sandbox.core.logging import setup_logging
from typosaurus_sandbox.research.client import RsearchClient
from typosaurus_sandbox.research.config import ResearchConfig
from typosaurus_sandbox.research.engine import ResearchEngine
logger = logging.getLogger(__name__)
def _enable_console_logging() -> None:
root = logging.getLogger()
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s"))
root.addHandler(console)
def main(argv: list[str] | None = None) -> None:
setup_logging()
_enable_console_logging()
parser = argparse.ArgumentParser(
prog="typosaurus-sandbox-research",
description="Exhaustive deep research over the rsearch API until closure",
)
parser.add_argument("subject", nargs="?", default="typosaurus sandbox", help="subject to research until closure")
args = parser.parse_args(argv)
config = ResearchConfig.load()
client = RsearchClient(config)
logger.info("research session starting subject=%r base_url=%s", args.subject, config.base_url)
report = asyncio.run(ResearchEngine(client=client).run(args.subject))
logger.info("research report %s", json.dumps(report.to_dict(), indent=2))
if __name__ == "__main__":
main()

205
src/typosaurus_sandbox/research/engine.py Normal file
View File

@ -0,0 +1,205 @@
# retoor <retoor@molodetz.nl>
import logging
from dataclasses import dataclass, field
from typing import AsyncIterator
from typosaurus_sandbox.research.client import RsearchClient
from typosaurus_sandbox.research.config import ResearchConfig
from typosaurus_sandbox.research.frontier import QueryFrontier
from typosaurus_sandbox.research.pipeline import PipelineReport, ResearchPipeline, WorkItem
logger = logging.getLogger(__name__)
@dataclass
class RoundSummary:
number: int = 0
items_processed: int = 0
requests_succeeded: int = 0
requests_failed: int = 0
cache_hits: int = 0
content_types: dict[str, int] = field(default_factory=dict)
new_urls: int = 0
new_queries: int = 0
new_contents: int = 0
closed: bool = False
def to_dict(self) -> dict[str, int | dict[str, int] | bool]:
return {
"number": self.number,
"items_processed": self.items_processed,
"requests_succeeded": self.requests_succeeded,
"requests_failed": self.requests_failed,
"cache_hits": self.cache_hits,
"content_types": self.content_types,
"new_urls": self.new_urls,
"new_queries": self.new_queries,
"new_contents": self.new_contents,
"closed": self.closed,
}
@dataclass
class ResearchReport:
subject: str
rounds: list[RoundSummary] = field(default_factory=list)
total_rounds: int = 0
queries_generated: int = 0
queries_enqueued: int = 0
queries_issued: int = 0
queries_duplicates_skipped: int = 0
urls_collected: int = 0
urls_duplicates_skipped: int = 0
contents_seen: int = 0
content_duplicates_skipped: int = 0
content_types: dict[str, int] = field(default_factory=dict)
requests_succeeded: int = 0
requests_failed: int = 0
cache_hits: int = 0
cache_misses: int = 0
closed: bool = False
def to_dict(self) -> dict[str, object]:
return {
"subject": self.subject,
"rounds": [round_summary.to_dict() for round_summary in self.rounds],
"total_rounds": self.total_rounds,
"queries_generated": self.queries_generated,
"queries_enqueued": self.queries_enqueued,
"queries_issued": self.queries_issued,
"queries_duplicates_skipped": self.queries_duplicates_skipped,
"urls_collected": self.urls_collected,
"urls_duplicates_skipped": self.urls_duplicates_skipped,
"contents_seen": self.contents_seen,
"content_duplicates_skipped": self.content_duplicates_skipped,
"content_types": self.content_types,
"requests_succeeded": self.requests_succeeded,
"requests_failed": self.requests_failed,
"cache_hits": self.cache_hits,
"cache_misses": self.cache_misses,
"closed": self.closed,
}
class ResearchEngine:
def __init__(
self,
client: RsearchClient | None = None,
frontier: QueryFrontier | None = None,
pipeline: ResearchPipeline | None = None,
) -> None:
self._client = client if client is not None else RsearchClient()
self._config: ResearchConfig = self._client.config
self._frontier = frontier if frontier is not None else QueryFrontier()
self._pipeline = pipeline if pipeline is not None else ResearchPipeline(self._client, self._frontier)
self._described_marker = 0
@property
def frontier(self) -> QueryFrontier:
return self._frontier
@property
def pipeline(self) -> ResearchPipeline:
return self._pipeline
async def _round_items(self, pending_queries: int, urls_to_describe: list[str]) -> AsyncIterator[WorkItem]:
for _ in range(pending_queries):
query = self._frontier.pop_query()
if query is None:
break
yield WorkItem("web", query)
yield WorkItem("images", query)
yield WorkItem("chat", query)
for url in urls_to_describe:
yield WorkItem("describe", url)
@staticmethod
def _round_summary(number: int, pipeline_report: PipelineReport) -> RoundSummary:
summary = RoundSummary(number=number, items_processed=len(pipeline_report.outcomes))
for outcome in pipeline_report.outcomes:
if outcome.success:
summary.requests_succeeded += 1
else:
summary.requests_failed += 1
if outcome.cache_hit:
summary.cache_hits += 1
kind = outcome.item.kind
summary.content_types[kind] = summary.content_types.get(kind, 0) + 1
return summary
async def run(self, subject: str) -> ResearchReport:
cleaned_subject = " ".join(subject.split())
if not cleaned_subject:
raise ValueError("research subject must not be empty")
self._frontier.seed(cleaned_subject)
report = ResearchReport(subject=cleaned_subject)
round_number = 0
while True:
round_start = self._frontier.snapshot()
pending_queries = round_start.queries_enqueued - round_start.queries_issued
urls_to_describe = self._frontier.urls_since(self._described_marker)
self._described_marker = round_start.urls_seen
if pending_queries == 0 and not urls_to_describe:
logger.info("research closed, no pending queries or urls after round %d", round_number)
break
round_number += 1
logger.info(
"round %d start pending_queries=%d urls_to_describe=%d",
round_number,
pending_queries,
len(urls_to_describe),
)
pipeline_report = await self._pipeline.run(self._round_items(pending_queries, urls_to_describe))
summary = self._round_summary(round_number, pipeline_report)
round_end = self._frontier.snapshot()
summary.new_urls = round_end.urls_seen - round_start.urls_seen
summary.new_queries = round_end.queries_enqueued - round_start.queries_enqueued
summary.new_contents = round_end.content_seen - round_start.content_seen
summary.closed = summary.new_urls == 0 and summary.new_queries == 0
report.rounds.append(summary)
logger.info(
"round %d finished new_urls=%d new_queries=%d new_contents=%d closed=%s",
round_number,
summary.new_urls,
summary.new_queries,
summary.new_contents,
summary.closed,
)
if summary.closed:
break
report.total_rounds = round_number
report.closed = True
self._finalize(report)
logger.info(
"research complete subject=%r rounds=%d queries_issued=%d urls_collected=%d contents_seen=%d cache_hits=%d",
report.subject,
report.total_rounds,
report.queries_issued,
report.urls_collected,
report.contents_seen,
report.cache_hits,
)
return report
def _finalize(self, report: ResearchReport) -> None:
stats = self._frontier.snapshot()
report.queries_generated = stats.queries_generated
report.queries_enqueued = stats.queries_enqueued
report.queries_issued = stats.queries_issued
report.queries_duplicates_skipped = stats.queries_duplicates_skipped
report.urls_collected = stats.urls_seen
report.urls_duplicates_skipped = stats.urls_duplicates_skipped
report.contents_seen = stats.content_seen
report.content_duplicates_skipped = stats.content_duplicates_skipped
total_items = 0
for summary in report.rounds:
total_items += summary.items_processed
report.requests_succeeded += summary.requests_succeeded
report.requests_failed += summary.requests_failed
report.cache_hits += summary.cache_hits
for kind, count in summary.content_types.items():
report.content_types[kind] = report.content_types.get(kind, 0) + count
report.cache_misses = total_items - report.cache_hits

6
src/typosaurus_sandbox/research/frontier.py
View File

@ -161,10 +161,15 @@ class QueryFrontier:
logger.debug("url duplicate skipped url=%s", normalized)
return False
self._seen_urls.add(normalized)
self._seen_url_order.append(normalized)
self._urls_seen += 1
logger.info("url registered url=%s", normalized)
return True
def urls_since(self, seen_count: int) -> list[str]:
with self._lock:
return list(self._seen_url_order[seen_count:])
def register_content(self, text: str) -> bool:
if not text.strip():
return False
@ -226,3 +231,4 @@ class QueryFrontier:
)