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 provider failures

Browse Source
This commit is contained in:
faligam
2026-09-06 18:32:42 +02:00
parent 9f6029b60a
commit e5b7d90436
9 changed files with 269 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -147,6 +147,7 @@ class StatusResponse(BaseModel):
search_count: int = 0
source_count: int = 0
claim_count: int = 0
error: str | None = None
is_completed: bool = False
is_running: bool = False
@@ -468,6 +469,8 @@ async def _run_pipeline(run: _ResearchRunState, request: ResearchRequest, budget
)
else:
run.state = ResearchRunState.FAILED.value
run.metadata["error"] = result.get("error", "Pipeline failed")
run.metadata["failed_step"] = result.get("failed_step", "unknown")
metrics.increment(C_RESEARCH_FAILED_TOTAL)
logger.warning(
"[pipeline] run_id=%s failed: %s",
@@ -519,6 +522,7 @@ async def get_research(research_id: str) -> StatusResponse:
search_count=len(run.search_results),
source_count=len(run.sources),
claim_count=len(run.claims),
error=run.metadata.get("error"),
is_completed=(run.state == ResearchRunState.COMPLETED.value),
is_running=run.state not in (
ResearchRunState.COMPLETED.value,
@@ -546,6 +550,7 @@ async def get_status(research_id: str) -> StatusResponse:
search_count=len(run.search_results),
source_count=len(run.sources),
claim_count=len(run.claims),
error=run.metadata.get("error"),
is_completed=(run.state == ResearchRunState.COMPLETED.value),
is_running=run.state not in (
ResearchRunState.COMPLETED.value,

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

@@ -208,8 +208,6 @@ class ResearchOrchestrator:
"""
logger.error("Research failed: %s (state=%s)", reason, self._state_machine.current_state.value)
# Der State Machine _apply erzwingt FAILED direkt,
# weil _mark_failed als "Sondertransition" gedacht ist.
if self._state_machine.current_state in (
ResearchRunState.COMPLETED,
ResearchRunState.FAILED,
@@ -217,8 +215,16 @@ class ResearchOrchestrator:
):
return False
self._state_machine.current_state = ResearchRunState.FAILED # type: ignore[assignment]
return True
transitioned = self._transition_to(ResearchRunState.FAILED.value)
if transitioned and self._run is not None:
self._run = self._run.model_copy(
update={
"state": ResearchRunState.FAILED.value,
"updated_at": datetime.now(timezone.utc),
"metadata": {**self._run.metadata, "error": reason},
}
)
return transitioned
# ---------------------------------------------------------------
# Private: Budget
@@ -297,14 +303,14 @@ class ResearchOrchestrator:
if self._config.searxng_base_url:
provider = SearXNGProvider(base_url=self._config.searxng_base_url)
providers.append(provider)
except ImportError:
pass
except ImportError as exc:
logger.error("SearXNG provider is unavailable: %s", exc)
except Exception as exc:
logger.warning("SearXNG provider init failed (%s) — search will return empty", exc)
logger.error("SearXNG provider initialization failed: %s", exc)
self._multi_search = MultiProviderSearch(providers=providers if providers else None)
self._multi_search = MultiProviderSearch(providers=providers)
if not providers:
logger.warning("No search providers configured — search will return empty results")
logger.error("No search providers configured")
return self._multi_search
def _get_llm_provider(self):
@@ -406,6 +412,7 @@ class ResearchOrchestrator:
"planning",
"searching",
"fetching",
"extracting",
"analyzing",
"comparing",
"synthesizing",
@@ -440,6 +447,7 @@ class ResearchOrchestrator:
"success": False,
"error": error_msg,
"failed_step": step_name,
"state": self._state_machine.current_state.value,
"report": None,
}
@@ -458,6 +466,7 @@ class ResearchOrchestrator:
"success": False,
"error": str(exc),
"failed_step": step_name,
"state": self._state_machine.current_state.value,
"report": None,
}
@@ -552,18 +561,7 @@ class ResearchOrchestrator:
except Exception as exc:
logger.error("Planning failed: %s", exc)
# Fallback: Create a minimal plan
fallback = self._get_fallback_plan()
self._plan = fallback
# Store error in _run metadata via new instance
if self._run is not None:
self._run = self._run.model_copy(
update={
"plan_error": str(exc),
"updated_at": datetime.utcnow(),
}
)
return {"success": True, "data": {"plan": fallback}, "plan": fallback, "warning": "Used fallback plan"}
return {"success": False, "error": f"Planning failed: {exc}", "data": {}}
async def _step_searching(self) -> dict[str, Any]:
"""SEARCHING: Führe Suchanfragen durch und sammle URLs."""
@@ -571,6 +569,12 @@ class ResearchOrchestrator:
try:
multi_search = self._get_multi_search()
if not multi_search.enabled_providers():
return {
"success": False,
"error": "No search provider is configured or available",
"data": {},
}
plan_queries = self._plan.get("queries", []) if self._plan else []
if not plan_queries:
@@ -615,9 +619,8 @@ class ResearchOrchestrator:
return {"success": True, "data": {"urls": self._search_results}, "url_count": len(self._search_results)}
except Exception as exc:
logger.warning("Searching failed (graceful fallback): %s — returning empty results", exc)
self._search_results = []
return {"success": True, "data": {"urls": []}, "url_count": 0, "warning": str(exc)}
logger.error("Searching failed: %s", exc)
return {"success": False, "error": f"Searching failed: {exc}", "data": {}}
async def _step_fetching(self) -> dict[str, Any]:
"""FETCHING: Crawle gesammelte URLs und extrahiere Inhalt."""
@@ -669,9 +672,8 @@ class ResearchOrchestrator:
}
except Exception as exc:
logger.warning("Fetching failed (graceful fallback): %s — returning empty sources", exc)
self._sources = []
return {"success": True, "data": {"sources": []}, "source_count": 0, "warning": str(exc)}
logger.error("Fetching failed: %s", exc)
return {"success": False, "error": f"Fetching failed: {exc}", "data": {}}
async def _step_extracting(self) -> dict[str, Any]:
"""EXTRACTING: Extrahiere Claims aus den extrahierten Quellen."""
@@ -717,9 +719,8 @@ class ResearchOrchestrator:
return {"success": True, "data": {"claims": [c.model_dump() for c in self._claims]}, "claim_count": len(self._claims)}
except Exception as exc:
logger.warning("Claim extraction failed (graceful fallback): %s", exc)
self._claims = []
return {"success": True, "data": {"claims": []}, "claim_count": 0, "warning": str(exc)}
logger.error("Claim extraction failed: %s", exc)
return {"success": False, "error": f"Claim extraction failed: {exc}", "data": {}}
async def _step_analyzing(self) -> dict[str, Any]:
"""ANALYZING: Platzhalter — Stage 6/7/8 werden später eingebunden.
@@ -1180,4 +1181,4 @@ class ResearchOrchestrator:
except Exception as exc:
logger.warning("Gap search execution failed: %s", exc)
return {"search_results": search_results, "claims": claims}
return {"search_results": search_results, "claims": claims}

2
src/nsct/orchestration/state.py
View File

@@ -39,7 +39,7 @@ class ResearchRunState(str, Enum):
# Übergangsmatrix: Quelle -> zulässige Ziele (frozenset für Immutable)
_VALID_TRANSITIONS: dict[ResearchRunState, FrozenSet[ResearchRunState]] = {
ResearchRunState.CREATED: frozenset(
(ResearchRunState.PLANNING,)
(ResearchRunState.PLANNING, ResearchRunState.FAILED)
),
ResearchRunState.PLANNING: frozenset(
(ResearchRunState.SEARCHING, ResearchRunState.FAILED)

10
src/nsct/providers/llm.py
View File

@@ -151,8 +151,14 @@ class _LLMProviderImpl(LLMProvider):
pool=5,
),
)
# OpenAI-compatible endpoints are commonly configured either as the
# service root or with the API version already included. Appending
# blindly turns an explicit ``.../v1`` into ``.../v1/v1``.
base_url = self._base_url.rstrip("/")
if not base_url.endswith("/v1"):
base_url += "/v1"
return AsyncOpenAI(
base_url=self._base_url.rstrip("/") + "/v1",
base_url=base_url,
api_key=self._api_key,
http_client=http_client,
)
@@ -354,4 +360,4 @@ def get_provider(config: AppSettings, metrics: ProviderMetrics) -> LLMProvider:
global _instance
if _instance is None:
_instance = _LLMProviderImpl(config, metrics)
return _instance
return _instance

82
src/nsct/providers/searxng.py Normal file
View File

@@ -0,0 +1,82 @@
"""SearXNG JSON search provider."""
from __future__ import annotations
from typing import Any
import httpx
from nsct.providers.abstract import NormalizedResult, SearchProvider
class SearXNGProvider(SearchProvider):
"""Search through a self-hosted SearXNG instance's JSON endpoint."""
_provider_name = "searxng"
def __init__(self, base_url: str, timeout_seconds: float = 15.0) -> None:
if not base_url.strip():
raise ValueError("SearXNG base URL must not be empty")
self._base_url = base_url.rstrip("/")
self._timeout_seconds = timeout_seconds
self._client: httpx.AsyncClient | None = None
@property
def _http_client(self) -> httpx.AsyncClient:
if self._client is None:
self._client = httpx.AsyncClient(timeout=httpx.Timeout(self._timeout_seconds))
return self._client
async def search(
self, query: str, language: str = "de", max_results: int = 10
) -> list[NormalizedResult]:
response = await self._http_client.get(
f"{self._base_url}/search",
params={"q": query, "format": "json", "language": language},
# The bundled SearXNG instance enables bot detection and rejects
# requests without a client IP header. This is the loopback IP of
# the API-to-SearXNG hop, not an end-user supplied value.
headers={"X-Forwarded-For": "127.0.0.1"},
)
response.raise_for_status()
payload = response.json()
raw_results = payload.get("results", [])
if not isinstance(raw_results, list):
raise ValueError("Invalid SearXNG response: results is not a list")
results: list[NormalizedResult] = []
for rank, item in enumerate(raw_results[:max_results], start=1):
if not isinstance(item, dict) or not isinstance(item.get("url"), str):
continue
url = item["url"]
if not url.startswith(("http://", "https://")):
continue
results.append(
NormalizedResult.from_raw(
provider_name=self._provider_name,
title=str(item.get("title") or url),
url=url,
snippet=str(item.get("content") or item.get("snippet") or ""),
rank=rank,
extra={"engine": item.get("engine"), "category": item.get("category")},
)
)
return results
async def get_metadata(self) -> dict[str, Any]:
return {
"name": self._provider_name,
"version": "0.1.0",
"provider": self.__class__.__name__,
"capabilities": ["web_search", "json_api"],
"requires_api_key": False,
"base_url": self._base_url,
"note": "Search ranking is NOT a trust indicator.",
}
async def health_check(self) -> bool:
try:
response = await self._http_client.get(f"{self._base_url}/healthz")
return response.is_success
except httpx.HTTPError:
return False