fix research result handoff and synthesis provenance

This commit is contained in:
faligam
2026-09-07 13:21:33 +02:00
parent 7b279dceea
commit dc285d80bd
5 changed files with 152 additions and 23 deletions

View File

@@ -561,9 +561,15 @@ async def _run_pipeline(run: _ResearchRunState, request: ResearchRequest, budget
if result.get("success"): if result.get("success"):
run.state = ResearchRunState.COMPLETED.value run.state = ResearchRunState.COMPLETED.value
run.report = result.get("report") run.report = result.get("report")
run.sources = result.get("sources", []) report_data = run.report if isinstance(run.report, dict) else {}
run.claims = result.get("report", {}).get("claims", []) # The orchestrator owns the collected sources. Earlier code read
run.evidence_scores = result.get("report", {}).get("evidence", []) # a non-existent top-level field and therefore displayed zero
# sources even though claims with valid provenance were present.
run.sources = result.get("sources", report_data.get("sources", []))
run.claims = result.get("claims", report_data.get("claims", []))
run.evidence_scores = result.get(
"evidence_scores", report_data.get("evidence", [])
)
metrics.increment(C_RESEARCH_COMPLETED_TOTAL) metrics.increment(C_RESEARCH_COMPLETED_TOTAL)
metrics.increment(C_SOURCES_FETCHED_TOTAL, len(run.sources)) metrics.increment(C_SOURCES_FETCHED_TOTAL, len(run.sources))
@@ -789,10 +795,31 @@ async def get_report(research_id: str) -> ReportResponse:
) )
report_data = run.report report_data = run.report
findings = [ raw_findings = report_data.get("findings", report_data.get("findings_json"))
FindingItem(**f) if isinstance(f, dict) else FindingItem(text=str(f)) if raw_findings is None:
for f in report_data.get("findings", report_data.get("findings_json", [])) # Stage 9 separates high-confidence findings and uncertainty. The
# REST contract exposes one display list, so retain both categories
# rather than silently rendering a blank report.
raw_findings = [
*report_data.get("confident_findings", []),
*report_data.get("uncertain_areas", []),
] ]
findings = []
for finding in raw_findings:
if not isinstance(finding, dict):
findings.append(FindingItem(text=str(finding)))
continue
if "text" in finding:
findings.append(FindingItem(**finding))
continue
source_id = finding.get("source_id")
findings.append(
FindingItem(
text=str(finding.get("claim_text", "")),
confidence=float(finding.get("confidence", 0.0)),
source_ids=[str(source_id)] if source_id else [],
)
)
return ReportResponse( return ReportResponse(
research_id=run.research_id, research_id=run.research_id,
@@ -801,9 +828,9 @@ async def get_report(research_id: str) -> ReportResponse:
findings=findings, findings=findings,
disagreements=report_data.get("disagreements", []), disagreements=report_data.get("disagreements", []),
uncertainties=report_data.get("uncertainties", []), uncertainties=report_data.get("uncertainties", []),
source_statistics=report_data.get("source_statistics", {}), source_statistics=report_data.get("source_statistics", {"total": len(run.sources)}),
methodology=report_data.get("methodology", ""), methodology=report_data.get("methodology", ""),
generated_at=report_data.get("generated_at", _now()), generated_at=str(report_data.get("generated_at", report_data.get("generation_timestamp", _now()))),
) )

View File

@@ -485,9 +485,25 @@ class ResearchOrchestrator:
self._transition_to("completed") self._transition_to("completed")
self._budget_tracker.record_time_elapsed() self._budget_tracker.record_time_elapsed()
# ``result`` is the synthesis result from the final pipeline step.
# Keep it instead of replacing it with the technical intermediate
# report below. The latter is useful provenance, but not the report
# contract consumed by the REST/UI layer.
synthesis_report = result.get("report") or result.get("data") or {}
if not isinstance(synthesis_report, dict):
synthesis_report = {}
report = { report = {
"success": True, "success": True,
"report": self._get_report(), "report": {
**synthesis_report,
"sources": self._sources,
"claims": [claim.model_dump(mode="json") for claim in self._claims],
},
"sources": self._sources,
"claims": [claim.model_dump(mode="json") for claim in self._claims],
"evidence_scores": list(
self._comparison_data.get("evidence_scores", {}).values()
) if isinstance(getattr(self, "_comparison_data", {}), dict) else [],
"state": self._state_machine.current_state.value, "state": self._state_machine.current_state.value,
"budget_usage": self._budget_tracker.get_usage(), "budget_usage": self._budget_tracker.get_usage(),
"plan": self._plan, "plan": self._plan,
@@ -897,7 +913,9 @@ class ResearchOrchestrator:
return self._fallback_synthesis() return self._fallback_synthesis()
# Claims als Dicts für SynthesisStage # Claims als Dicts für SynthesisStage
claims_dicts = [c.model_dump() for c in self._claims] # Stage 9 serializes this package for the LLM. JSON mode is
# essential: the claim model contains UUIDs and datetimes.
claims_dicts = [c.model_dump(mode="json") for c in self._claims]
stage = SynthesisStage( stage = SynthesisStage(
research_run_id=self._run.id if self._run else uuid4(), research_run_id=self._run.id if self._run else uuid4(),
@@ -906,7 +924,7 @@ class ResearchOrchestrator:
claims=claims_dicts, claims=claims_dicts,
) )
result = await stage.execute() result = await stage.execute(topic=self._query)
if result.success: if result.success:
logger.info("Synthesis complete: report generated successfully") logger.info("Synthesis complete: report generated successfully")
@@ -918,8 +936,17 @@ class ResearchOrchestrator:
"report": report, "report": report,
} }
else: else:
logger.warning("Synthesis returned success=False — generating fallback") # Stage 9 supplies a provenance-preserving uncertainty report
return self._fallback_synthesis_data(result.errors) # on failure. It is a successful degraded pipeline outcome,
# not an empty technical fallback.
logger.warning("Synthesis returned success=False — returning uncertainty report")
return {
"success": True,
"data": result.data,
"report": result.data,
"warning": "Synthesis degraded to uncertainty report",
"synthesis_errors": result.errors,
}
except Exception as exc: except Exception as exc:
logger.error("Synthesis failed: %s — generating fallback", exc) logger.error("Synthesis failed: %s — generating fallback", exc)
@@ -984,13 +1011,13 @@ class ResearchOrchestrator:
def _get_report(self) -> dict[str, Any]: def _get_report(self) -> dict[str, Any]:
"""Erzeuge den finalen Report aus allen Zwischenspeichern.""" """Erzeuge den finalen Report aus allen Zwischenspeichern."""
report: dict[str, Any] = { report: dict[str, Any] = {
"run_id": self._run.id if self._run else None, "run_id": str(self._run.id) if self._run else None,
"query": self._query, "query": self._query,
"state": self._state_machine.current_state.value, "state": self._state_machine.current_state.value,
"plan": self._plan, "plan": self._plan,
"search_results": self._search_results, "search_results": self._search_results,
"sources": self._sources, "sources": self._sources,
"claims": [c.model_dump() for c in self._claims], "claims": [c.model_dump(mode="json") for c in self._claims],
"claim_count": len(self._claims), "claim_count": len(self._claims),
"source_count": len(self._sources), "source_count": len(self._sources),
"budget_usage": self._budget_tracker.get_usage(), "budget_usage": self._budget_tracker.get_usage(),

View File

@@ -320,13 +320,16 @@ def _build_evidence_package_json(
# Index scores by claim_id for fast lookup # Index scores by claim_id for fast lookup
score_map: dict[str, dict[str, Any]] = {} score_map: dict[str, dict[str, Any]] = {}
for sc in scores: for sc in scores:
cid = sc.get("claim_id", "") cid = str(sc.get("claim_id", sc.get("id", "")))
if cid: if cid:
score_map[cid] = sc score_map[cid] = sc
evidence_items = [] evidence_items = []
for claim in claims: for claim in claims:
claim_id = claim.get("claim_id", "") # Stage 5's canonical model field is ``id`` while older stage tests
# and score records use ``claim_id``. Normalize both at the boundary
# so an extracted claim never loses its provenance in Stage 9.
claim_id = str(claim.get("claim_id") or claim.get("id", ""))
score = score_map.get(claim_id, {}) score = score_map.get(claim_id, {})
# Build evidence entry with full provenance # Build evidence entry with full provenance
@@ -335,7 +338,7 @@ def _build_evidence_package_json(
"claim_text": claim.get("claim_text", ""), "claim_text": claim.get("claim_text", ""),
"claim_type": claim.get("claim_type", "claim"), "claim_type": claim.get("claim_type", "claim"),
"confidence": claim.get("confidence", 1.0), "confidence": claim.get("confidence", 1.0),
"source_id": claim.get("source_id", ""), "source_id": str(claim.get("source_id", "")),
"source_url": claim.get("source_url", ""), "source_url": claim.get("source_url", ""),
"source_title": claim.get("source_title"), "source_title": claim.get("source_title"),
"evidence_span": claim.get("evidence_span"), "evidence_span": claim.get("evidence_span"),
@@ -596,7 +599,7 @@ class SynthesisStage(BaseStage):
return StageResult( return StageResult(
success=True, success=True,
data=report.model_dump(), data=report.model_dump(mode="json"),
stage=self, stage=self,
) )
@@ -732,7 +735,10 @@ class Stage9Synthesis:
async def run(self) -> dict[str, Any]: async def run(self) -> dict[str, Any]:
"""Backward-compatible async .run() method.""" """Backward-compatible async .run() method."""
result = await self._stage.execute() result = await self._stage.execute()
if not result.success: # A failed LLM call can still yield Stage 9's valid, provenance-
# preserving uncertainty report. Only propagate a failure when no
# usable report was produced (for example, when there are no claims).
if not result.success and not result.data:
raise RuntimeError( raise RuntimeError(
"Stage 9 synthesis failed: " + "; ".join(result.errors) "Stage 9 synthesis failed: " + "; ".join(result.errors)
) )

View File

@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import json
from uuid import uuid4 from uuid import uuid4
import httpx import httpx
@@ -18,6 +19,8 @@ from nsct.providers.metrics import ProviderMetrics
from nsct.providers.searxng import SearXNGProvider from nsct.providers.searxng import SearXNGProvider
from nsct.crawler.normalize import NormalizedDocument from nsct.crawler.normalize import NormalizedDocument
from nsct.api.rest_research import DEPTH_CONFIGS from nsct.api.rest_research import DEPTH_CONFIGS
from nsct.models.claim import Claim, ClaimType
from nsct.stages.stage9_synthesis import _build_evidence_package_json
def _config(*, searxng_base_url: str | None = "http://searxng:8080") -> AppSettings: def _config(*, searxng_base_url: str | None = "http://searxng:8080") -> AppSettings:
@@ -206,6 +209,59 @@ def test_depth_time_budgets_allow_local_35b_inference() -> None:
assert DEPTH_CONFIGS["deep"].max_research_duration_seconds == 7_200 assert DEPTH_CONFIGS["deep"].max_research_duration_seconds == 7_200
def test_synthesis_package_keeps_stage5_uuid_claim_provenance() -> None:
"""Stage 9 must accept the actual, UUID-bearing Stage 5 claim model."""
source_id = uuid4()
claim = Claim(
research_run_id=uuid4(),
source_id=source_id,
claim_text="Eine überprüfbare Behauptung.",
evidence_span="Die belegende Passage.",
claim_type=ClaimType.CLAIM,
source_url="https://example.org/source",
)
package = json.loads(_build_evidence_package_json([claim.model_dump()], [], "Thema"))
assert package["evidence"][0]["claim_id"] == str(claim.id)
assert package["evidence"][0]["source_id"] == str(source_id)
assert package["sources"][0]["source_id"] == str(source_id)
def test_orchestrator_synthesis_returns_a_report_for_uuid_claims() -> None:
"""The production hand-off to Stage 9 must not trigger UUID JSON errors."""
async def run() -> None:
orchestrator = ResearchOrchestrator(_config(), uuid4(), "Testthema")
await orchestrator.start()
orchestrator._state_machine = StateMachine(ResearchRunState.COMPARING)
orchestrator._claims = [
Claim(
research_run_id=uuid4(),
source_id=uuid4(),
claim_text="Eine überprüfbare Behauptung.",
evidence_span="Die belegende Passage.",
claim_type=ClaimType.CLAIM,
source_url="https://example.org/source",
)
]
provider = Mock()
provider.complete = AsyncMock(return_value=json.dumps({
"summary": "Neutraler Bericht.",
"confident_findings": [],
"uncertain_areas": [],
"contradictions": [],
}))
orchestrator._llm_provider = provider
response = await orchestrator._step_synthesizing()
assert response["success"] is True
assert response["report"]["summary"] == "Neutraler Bericht."
provider.complete.assert_awaited_once()
asyncio.run(run())
def test_start_does_not_charge_a_phantom_planner_request() -> None: def test_start_does_not_charge_a_phantom_planner_request() -> None:
async def run() -> None: async def run() -> None:
orchestrator = ResearchOrchestrator(_config(), uuid4(), "test") orchestrator = ResearchOrchestrator(_config(), uuid4(), "test")

View File

@@ -334,7 +334,18 @@ def test_detail_endpoints_serialize_pipeline_uuid_identifiers(client: TestClient
} }
], ],
evidence_scores=[{"claim_id": claim_id, "research_run_id": uuid4()}], evidence_scores=[{"claim_id": claim_id, "research_run_id": uuid4()}],
report={"summary": "Zusammenfassung", "findings": [], "methodology": "Methodik"}, report={
"summary": "Zusammenfassung",
"confident_findings": [
{
"claim_text": "Eine überprüfbare Behauptung.",
"confidence": 0.9,
"source_id": source_id,
}
],
"uncertain_areas": [],
"methodology": "Methodik",
},
) )
sources = client.get(f"/v1/research/{research_id}/sources") sources = client.get(f"/v1/research/{research_id}/sources")
@@ -348,6 +359,8 @@ def test_detail_endpoints_serialize_pipeline_uuid_identifiers(client: TestClient
assert claims.json()["claims"][0]["source_id"] == str(source_id) assert claims.json()["claims"][0]["source_id"] == str(source_id)
assert evidence.json()["evidence"][0]["claim_id"] == str(claim_id) assert evidence.json()["evidence"][0]["claim_id"] == str(claim_id)
assert report.json()["methodology"] == "Methodik" assert report.json()["methodology"] == "Methodik"
assert report.json()["findings"][0]["text"] == "Eine überprüfbare Behauptung."
assert report.json()["findings"][0]["source_ids"] == [str(source_id)]
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------