opencode/NSCT---Neutral-Search-Crawler-Tool
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix research pipeline runtime failures

Browse Source
This commit is contained in:
faligam
2026-09-07 11:16:36 +02:00
parent cf10cd9636
commit a5324d3971
9 changed files with 184 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/nsct/api/rest_research.py
View File

@@ -66,7 +66,7 @@ DEPTH_CONFIGS: dict[str, DepthConfig] = {
max_pages_per_domain=2,
max_total_download_bytes=500_000,
max_llm_requests=15,
max_research_duration_seconds=120,
max_research_duration_seconds=300,
max_context_per_llm_call=16_000,
),
"normal": DepthConfig(

3
src/nsct/crawler/fetcher.py
View File

@@ -98,6 +98,7 @@ class AsyncFetcher:
connect=float(os.environ.get("NSCT_CRAWLER_CONNECT_TIMEOUT", "10")),
read=float(os.environ.get("NSCT_CRAWLER_READ_TIMEOUT", "30")),
write=float(os.environ.get("NSCT_CRAWLER_WRITE_TIMEOUT", "10")),
pool=float(os.environ.get("NSCT_CRAWLER_POOL_TIMEOUT", "10")),
)
self._client = httpx.AsyncClient(
timeout=timeout_config,
@@ -304,4 +305,4 @@ class AsyncFetcher:
@staticmethod
def compute_content_hash(content: bytes) -> str:
"""Compute SHA256 hash of content."""
return hashlib.sha256(content).hexdigest()
return hashlib.sha256(content).hexdigest()

12
src/nsct/crawler/manager.py
View File

@@ -36,7 +36,7 @@ class CrawlerManager:
# Public API
# ------------------------------------------------------------------
async def fetch_and_extract(self, url: str) -> NormalizedDocument:
async def fetch_and_extract(self, url: str, max_size: int = 0) -> NormalizedDocument:
"""Fetch a single URL and extract main content.
Pipeline:
@@ -73,7 +73,7 @@ class CrawlerManager:
logger.warning("DNS check warning for %s: %s", url, exc)
# 3. HTTP fetch
result = await self.fetcher.fetch(url)
result = await self.fetcher.fetch(url, max_size=max_size)
if result.status in (FetchStatus.BLOCKED, FetchStatus.ERROR, FetchStatus.TIMEOUT,
FetchStatus.SIZE_LIMIT_EXCEEDED, FetchStatus.CONTENT_TYPE_BLOCKED):
@@ -86,7 +86,7 @@ class CrawlerManager:
# 4. Content-based extraction
text = ""
extraction_tool = ""
metadata: dict[str, Any] = {}
metadata: dict[str, Any] = {"download_bytes": len(content)}
links: list[str] = []
title = ""
@@ -131,7 +131,7 @@ class CrawlerManager:
return doc
async def fetch_and_extract_many(self, urls: list[str]) -> list[NormalizedDocument]:
async def fetch_and_extract_many(self, urls: list[str], max_size: int = 0) -> list[NormalizedDocument]:
"""Fetch and extract multiple URLs in parallel.
Each URL is processed independently — failures do not block others.
@@ -143,7 +143,7 @@ class CrawlerManager:
List of NormalizedDocument, one per input URL (in order).
Failed URLs produce error documents with error info in metadata.
"""
tasks = [self.fetch_and_extract(url) for url in urls]
tasks = [self.fetch_and_extract(url, max_size=max_size) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
final_docs: list[NormalizedDocument] = []
@@ -262,4 +262,4 @@ class CrawlerManager:
if lang_match:
meta["language"] = lang_match.group(1)
return meta
return meta

5
src/nsct/orchestration/budget.py
View File

@@ -160,7 +160,10 @@ class BudgetTracker:
counter_name = field.replace("max_", "", 1)
limit = getattr(self._config, field)
usage = self._counters.get(counter_name, 0)
if limit > 0 and usage >= limit:
# Limits are inclusive: exactly max_sources or exactly the byte
# budget is permitted; only use beyond the configured maximum
# exhausts the budget.
if limit > 0 and usage > limit:
exceeded.append(field)
logger.warning(
"Budget limit exceeded: %s (%.2f / %s)",

57
src/nsct/orchestration/orchestrator.py
View File

@@ -11,6 +11,7 @@ import asyncio
import json
import logging
import time
from collections.abc import Mapping
from datetime import datetime, timezone
from typing import Any
from uuid import UUID, uuid4
@@ -603,14 +604,30 @@ class ResearchOrchestrator:
seen_urls = set()
self._search_results = []
for r in all_results:
url = r.url if hasattr(r, "url") else r.get("url", "")
# SearchProvider returns NormalizedResult instances. Keep the
# mapping branch for test doubles and legacy providers, but do
# not call ``.get()`` on a Pydantic model when an optional
# field such as ``snippet`` is empty.
if hasattr(r, "url"):
url = r.url
title = r.title
snippet = r.snippet
provider = r.provider
elif isinstance(r, Mapping):
url = r.get("url", "")
title = r.get("title", "")
snippet = r.get("snippet", "")
provider = r.get("provider", "")
else:
logger.warning("Ignoring unsupported search result type: %s", type(r).__name__)
continue
if url and url not in seen_urls:
seen_urls.add(url)
result_dict = {
"url": url,
"title": getattr(r, "title", "") or r.get("title", ""),
"snippet": getattr(r, "snippet", "") or r.get("snippet", ""),
"provider": getattr(r, "provider", "") or r.get("provider", ""),
"title": title,
"snippet": snippet,
"provider": provider,
}
self._search_results.append(result_dict)
@@ -635,11 +652,31 @@ class ResearchOrchestrator:
crawler = self._get_crawler()
urls = [r["url"] for r in self._search_results]
# Budget: Abschätzung der Download-Größe
self._budget_tracker.increment_download_bytes(len(urls) * 500_000) # ~500KB pro Seite
# Restrict the batch before fetching. Charging each URL a
# fictitious 500 KB exhausted a quick run before any bytes had
# actually been downloaded.
usage = self._budget_tracker.get_usage()
remaining_sources = self._budget_config.max_sources - int(
usage["max_sources"]["usage"]
)
urls = urls[:max(0, remaining_sources)]
if not urls:
return {
"success": False,
"error": "Source budget exhausted before fetching",
"data": {},
}
remaining_download_bytes = self._budget_config.max_total_download_bytes - int(
usage["max_total_download_bytes"]["usage"]
)
max_size_per_url = 0
if self._budget_config.max_total_download_bytes > 0:
max_size_per_url = max(1, remaining_download_bytes // len(urls))
self._budget_tracker.increment_sources(len(urls))
docs = await crawler.fetch_and_extract_many(urls)
docs = await crawler.fetch_and_extract_many(urls, max_size=max_size_per_url)
self._budget_tracker.record_time_elapsed()
# In sources-Dicts umwandeln
@@ -657,9 +694,9 @@ class ResearchOrchestrator:
}
self._sources.append(src)
# Track bytes
content_len = len(doc.text or "")
self._budget_tracker.increment_download_bytes(content_len)
# Account for response bytes, not only extracted text.
downloaded_bytes = int(doc.metadata.get("download_bytes", len(doc.text.encode("utf-8"))))
self._budget_tracker.increment_download_bytes(downloaded_bytes)
successful = [s for s in self._sources if not s.get("error")]
metrics.increment(C_SOURCES_FETCHED_TOTAL, len(self._sources))