tests/test_reasoning_empty_report.py
Rama main · réplica de lectura del repositorio original.
"""Reasoning models that use up the whole token budget.
Reproduces a Kimi-like reasoning stream: the model delivers only
`reasoning_content`, uses up the whole token budget with it and ends the
stream with finish_reason="length" — without ever sending visible text.
"""
from src.llm.client import THINKING_PLACEHOLDER
import asyncio
import types
import tests.conftest # noqa: F401 (dependency stubs)
from src.config import LLMConfig
from src.llm.client import (
EmptyLLMResponseError, LLMClient, _reasoning_of,
)
def check(name, cond, detail=""):
assert cond, f"{name}" + (f" [{detail}]" if detail else "")
# ── Fake objects that mimic the OpenAI response structure ────────
class Delta:
def __init__(self, content=None, reasoning_content=None):
self.content = content
self.reasoning_content = reasoning_content
class Choice:
def __init__(self, delta=None, message=None, finish_reason=None):
self.delta = delta
self.message = message
self.finish_reason = finish_reason
class Chunk:
def __init__(self, choice):
self.choices = [choice]
class FakeStream:
def __init__(self, chunks):
self._chunks = chunks
def __aiter__(self):
async def gen():
for c in self._chunks:
yield c
return gen()
async def close(self):
pass
class FakeCompletions:
"""Record the max_tokens of every call (for the escalation)."""
def __init__(self, script):
self.script = list(script)
self.calls = []
async def create(self, **kwargs):
self.calls.append(kwargs.get("max_tokens"))
return self.script.pop(0)
def wire(client, script):
comp = FakeCompletions(script)
client.client = types.SimpleNamespace(
chat=types.SimpleNamespace(completions=comp)
)
return comp
def make_client(max_output_tokens=32768):
cfg = LLMConfig(model_name="kimi-25",
max_output_tokens=max_output_tokens)
return LLMClient(cfg)
async def collect(gen):
out = []
async for chunk in gen:
out.append(chunk)
return out
# ── 1. The reasoning field is recognised at all ─────────────────────
def test_reasoning_field_detection():
check("_reasoning_of reads reasoning_content",
_reasoning_of(Delta(reasoning_content="denk denk")) == "denk denk")
check("_reasoning_of ignores plain content",
_reasoning_of(Delta(content="Text")) == "")
# ── 2. A stream with reasoning only ────────────────────
async def test_stream_reasoning_only():
client = make_client()
# budget exhausted -> every attempt ends the same way
def budget_burn():
return FakeStream([
Chunk(Choice(delta=Delta(reasoning_content="x" * 400)))
for _ in range(10)
] + [Chunk(Choice(delta=Delta(), finish_reason="length"))])
comp = wire(client, [budget_burn() for _ in range(4)])
raised = None
chunks = []
try:
async for c in client.stream([{"role": "user", "content": "q"}],
max_tokens=16384):
chunks.append(c)
except EmptyLLMResponseError as e:
raised = e
check("a stream without visible text raises EmptyLLMResponseError",
raised is not None,
f"finish_reason={getattr(raised, 'finish_reason', None)}")
check("reasoning produces NO UI updates",
chunks == [], f"{len(chunks)} Chunks ausgegeben")
check("the budget was escalated once",
comp.calls[:2] == [16384, 32768], f"calls={comp.calls}")
def test_stream_reasoning_only_sync():
asyncio.run(test_stream_reasoning_only())
# ── 3. Regression: a normal stream keeps working ────────────
async def test_stream_ok():
client = make_client()
wire(client, [FakeStream([
Chunk(Choice(delta=Delta(reasoning_content="ueberlege..."))),
Chunk(Choice(delta=Delta(content="# Bericht\n"))),
Chunk(Choice(delta=Delta(content="Inhalt."),
finish_reason="stop")),
])])
chunks = await collect(client.stream([{"role": "user", "content": "q"}]))
check("reasoning + text -> only the text arrives",
chunks[-1] == "# Bericht\nInhalt.", repr(chunks[-1]))
check("no placeholder spam during reasoning",
not any(THINKING_PLACEHOLDER in c for c in chunks),
f"{len(chunks)} Chunks")
def test_stream_ok_sync():
asyncio.run(test_stream_ok())
async def test_unclosed_think_tag():
"""The placeholder from an open <think> block is not content."""
client = make_client(max_output_tokens=4096)
def only_think():
return FakeStream([
Chunk(Choice(delta=Delta(content="<think>gruebel"))),
Chunk(Choice(delta=Delta(content=" weiter"),
finish_reason="length")),
])
wire(client, [only_think() for _ in range(4)])
raised = None
try:
await collect(client.stream([{"role": "user", "content": "q"}],
max_tokens=4096))
except EmptyLLMResponseError as e:
raised = e
check("an open <think> block does not count as visible text",
raised is not None)
def test_unclosed_think_tag_sync():
asyncio.run(test_unclosed_think_tag())
# ── 4. complete(): escalation rescues the map call ─────────────────
async def test_complete_escalation():
client = make_client()
empty = types.SimpleNamespace(
choices=[Choice(message=types.SimpleNamespace(
content="", reasoning_content="y" * 5000),
finish_reason="length")],
usage=None)
full = types.SimpleNamespace(
choices=[Choice(message=types.SimpleNamespace(
content="Die Antwort.", reasoning_content="y" * 200),
finish_reason="stop")],
usage=None)
comp = wire(client, [empty, full])
out = await client.complete([{"role": "user", "content": "q"}],
max_tokens=4096)
check("complete() escalates the budget and then returns text",
out == "Die Antwort.", f"calls={comp.calls}")
def test_complete_escalation_sync():
asyncio.run(test_complete_escalation())
async def test_complete_gives_up_loudly():
client = make_client(max_output_tokens=4096) # no leeway
empty = types.SimpleNamespace(
choices=[Choice(message=types.SimpleNamespace(
content="", reasoning_content="y" * 5000),
finish_reason="length")],
usage=None)
wire(client, [empty])
try:
await client.complete([{"role": "user", "content": "q"}],
max_tokens=4096)
check("complete() without leeway raises an exception", False)
except EmptyLLMResponseError:
check("complete() without leeway raises an exception", True)
def test_complete_gives_up_loudly_sync():
asyncio.run(test_complete_gives_up_loudly())
# ── 5. Fallback helpers in the orchestrator ────────────────────────────
from src.pipeline.orchestrator import (
_report_from_map_answers, _strip_thinking_placeholder,
)
def test_orchestrator_fallback_helpers():
check("the placeholder is discarded as a report",
_strip_thinking_placeholder(THINKING_PLACEHOLDER) == ""
and _strip_thinking_placeholder("🤔 *Denkt nach...*") == "")
check("real text stays untouched",
_strip_thinking_placeholder("# Bericht\n\nInhalt")
== "# Bericht\n\nInhalt")
body = _report_from_map_answers([
{"question_id": "F1", "question": "Wer ist X?", "answer": "X ist Y."},
{"question_id": "F2", "question": "Leer?", "answer": " "},
{"question_id": "F3", "question": "Rolle?", "answer": "Leitung."},
])
check("the emergency report uses only filled map answers",
body.count("##") == 2 and "X ist Y." in body, repr(body[:60]))
check("no pseudo-report without map answers",
_report_from_map_answers([]) == "")
# ── 6. Node guards ──────────────────────────────────────────────
from src.pipeline.dag_nodes import (
_MIN_REPORT_CHARS, ReportRevisionNode, ReportWarningBannerNode,
)
class Ctx:
def __init__(self, **kw):
self.final_report = ""
self.factoid_verifications = []
self.report_quality = None
self.query_fulfillment = None
self.synthesis_degraded = ""
self.__dict__.update(kw)
def test_revision_guards():
# A degenerate report: 260 characters, 8 contradicted statements
degenerate = Ctx(
final_report="# Jonas Brenner\n\n> Anfrage\n\n---\n\n*Erstellt*"
+ " " * 200,
factoid_verifications=[
{"verified": "false", "confidence": 0.95, "factoid": f"A{i}"}
for i in range(8)
],
)
ok, reason = ReportRevisionNode(llm=None).applies_to(degenerate)
check("the revision leaves a degenerate report alone", ok is False, reason)
gesund = Ctx(final_report="x" * 5000,
factoid_verifications=degenerate.factoid_verifications)
ok, _ = ReportRevisionNode(llm=None).applies_to(gesund)
check("the revision still runs for a normal report", ok is True)
async def test_banner():
node = ReportWarningBannerNode()
ctx = Ctx(
final_report="# Titel\n\nInhalt",
report_quality={"passed": False, "rating": "poor",
"issues": [{"description": "Keine Fakten"}]},
query_fulfillment={"fulfilled": False, "assessment": "nur Metadaten",
"rework": "Synthese wiederholen"},
synthesis_degraded="Reasoning-Budget erschoepft.",
)
ok, _ = node.applies_to(ctx)
check("the banner triggers on complaints", ok is True)
res = await node.run(ctx)
check("the banner reports all three findings", res["problems"] == 3,
str(res))
check("the findings are in the report",
"Quality check not passed" in ctx.final_report
and "Reasoning-Budget" in ctx.final_report)
clean = Ctx(final_report="# Titel\n\nInhalt",
report_quality={"passed": True},
query_fulfillment={"fulfilled": True})
ok, reason = node.applies_to(clean)
check("no banner for a clean run", ok is False, reason)
def test_banner_sync():
asyncio.run(test_banner())
def test_min_report_threshold():
check("the threshold covers the degenerate report (260 characters)",
_MIN_REPORT_CHARS > 260, f"_MIN_REPORT_CHARS={_MIN_REPORT_CHARS}")