diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 52c7762d9b..f097dee742 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -1211,6 +1211,9 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]: parts: List[str] = [] def _add(text) -> None: + from agent.message_content import flatten_message_text + + text = flatten_message_text(text, sep="") if text and text not in parts: parts.append(text) _add(getattr(assistant_message, "reasoning", None)) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index fd6ecd4672..7f587a9752 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -6422,12 +6422,15 @@ class _ChatStreamAccumulator: if delta is None: return made_progress = False - piece = getattr(delta, "content", None) + from agent.message_content import flatten_message_text + + piece = flatten_message_text(getattr(delta, "content", None), sep="") if piece: self.content_parts.append(piece) made_progress = True reasoning_piece = getattr(delta, "reasoning", None) or getattr(delta, "reasoning_content", None) - if reasoning_piece and isinstance(reasoning_piece, str): + reasoning_piece = flatten_message_text(reasoning_piece, sep="") + if reasoning_piece: self.reasoning_parts.append(reasoning_piece) made_progress = True # Evaluate both unconditionally: they accumulate state, not just progress. diff --git a/agent/reasoning_summaries.py b/agent/reasoning_summaries.py index f58d150d7d..ef6c7f7d89 100644 --- a/agent/reasoning_summaries.py +++ b/agent/reasoning_summaries.py @@ -10,10 +10,14 @@ the blank-line join Hermes' own Responses adapter does. from __future__ import annotations +from typing import Any + +from agent.message_content import flatten_message_text + __all__ = ["separate_glued_reasoning_blocks"] -def separate_glued_reasoning_blocks(previous: str, delta: str) -> str: +def separate_glued_reasoning_blocks(previous: str, delta: Any) -> str: """Return *delta*, prefixed with a paragraph break when it glues onto *previous*. A break is inserted when *delta* opens a *closed* bold heading and *previous* is mid-line @@ -21,5 +25,7 @@ def separate_glued_reasoning_blocks(previous: str, delta: str) -> str: alone: its deltas carry their own whitespace, and a fragment that merely opens emphasis (``**`` alone) is not a part boundary — summary parts carry the whole heading in one delta. """ + # Relays also emit content-part lists/dicts; fragments carry their own whitespace. + delta = flatten_message_text(delta, sep="") glued = previous and delta and not previous[-1].isspace() and delta.startswith("**") and "**" in delta[2:] return f"\n\n{delta}" if glued else delta diff --git a/evals/providers/reasoning_shapes.py b/evals/providers/reasoning_shapes.py new file mode 100644 index 0000000000..7d9851e4b8 --- /dev/null +++ b/evals/providers/reasoning_shapes.py @@ -0,0 +1,169 @@ +"""Local HTTP/SDK reasoning-shape probe; no vendor inference or credentials. + +Run: python evals/providers/reasoning_shapes.py --output /tmp/reasoning.json +Run the same file in a fresh interpreter on base and fix checkouts for A/B. +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import sys +import tempfile +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + + +CASES = { + "string": ("**First**", "**Second**", "**First****Second**"), + "text-parts": ([{"type": "text", "text": "**First**"}], + [{"type": "text", "text": "**Second**"}], "**First****Second**"), + "split-fragments": (["Hel", "lo"], [" wor", "ld"], "Hello world"), + "dict": ({"type": "text", "text": "Hello"}, {"text": " world"}, "Hello world"), + "empty": ([], None, ""), + "nontext-parts": ([{"type": "image_url", "image_url": {"url": "ignored"}}], [], ""), +} + + +def run_matrix(surfaces): + from openai import AsyncOpenAI, OpenAI + from run_agent import AIAgent + from agent.agent_runtime_helpers import extract_reasoning + from agent.auxiliary_client import _aggregate_chat_stream, _aggregate_chat_stream_async + from agent.chat_completion_helpers import interruptible_streaming_api_call + from agent.chat_completion_helpers_relay import RelayChatAccumulator + + active = {"message": {}, "deltas": [], "answer": "LOCAL_CAPTURE_OK"} + requests = [] + + class Handler(BaseHTTPRequestHandler): + def log_message(self, format, *args): + pass + + def do_POST(self): + request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + requests.append({"path": self.path, "body": request}) + self.send_response(200) + stream = request.get("stream", False) + self.send_header("Content-Type", "text/event-stream" if stream else "application/json") + self.end_headers() + common = {"id": "local", "created": 1, "model": "probe"} + if stream: + for delta in active["deltas"] + [{"content": active["answer"]}]: + chunk = {**common, "object": "chat.completion.chunk", "choices": [ + {"index": 0, "delta": delta, "finish_reason": None}]} + self.wfile.write(("data: " + json.dumps(chunk) + "\n\n").encode()) + chunk = {**common, "object": "chat.completion.chunk", "choices": [ + {"index": 0, "delta": {}, "finish_reason": "stop"}]} + self.wfile.write(("data: " + json.dumps(chunk) + "\n\ndata: [DONE]\n\n").encode()) + else: + message = {"role": "assistant", "content": "LOCAL_CAPTURE_OK", **active["message"]} + response = {**common, "object": "chat.completion", "choices": [ + {"index": 0, "message": message, "finish_reason": "stop"}]} + self.wfile.write(json.dumps(response).encode()) + self.wfile.flush() + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + url = f"http://127.0.0.1:{server.server_port}/v1" + rows = [] + client = OpenAI(base_url=url, api_key="local-fixture", max_retries=0) + agent = AIAgent(base_url=url, api_key="local-fixture", provider="custom", model="probe", + api_mode="chat_completions", enabled_toolsets=[], quiet_mode=True, + skip_memory=True, skip_context_files=True, skip_background_review=True) + kwargs = {"model": "probe", "messages": [{"role": "user", "content": "probe"}]} + + async def aux_async(): + async with AsyncOpenAI(base_url=url, api_key="local-fixture", max_retries=0) as async_client: + stream = await async_client.chat.completions.create(**kwargs, stream=True) + return await _aggregate_chat_stream_async(stream, model="probe") + + try: + for surface in surfaces: + fields = ("reasoning", "reasoning_content") + if surface == "nonstream": + fields += tuple(f"reasoning_details.{key}" for key in ("summary", "thinking", "content", "text")) + for field in fields: + for name, (first, second, plain) in CASES.items(): + active["deltas"] = [{field: first}, {field: second}] + active["answer"] = ([{"type": "text", "text": "LOCAL_"}, "CAPTURE_OK"] + if name == "split-fragments" else "LOCAL_CAPTURE_OK") + # Completed responses exercise one field containing the same fragments. + active["message"] = ({"reasoning_details": [{field.split(".")[1]: first}]} + if "." in field else {field: first}) + callbacks = [] + agent.reasoning_callback = callbacks.append + expected = plain + if surface in ("main", "relay") and name in ("string", "text-parts"): + expected = "**First**\n\n**Second**" + if surface == "nonstream": + expected = {"string": "**First**", "text-parts": "**First**", + "split-fragments": "Hello", "dict": "Hello", + "empty": "", "nontext-parts": ""}[name] + row = {"surface": surface, "field": field, "case": name, "expected": expected} + try: + if surface == "main": + response = interruptible_streaming_api_call(agent, dict(kwargs)) + message = response.choices[0].message + reasoning = extract_reasoning(agent, message) or "" + row["callbacks"] = callbacks + elif surface == "relay": + acc = RelayChatAccumulator() + with client.chat.completions.create(**kwargs, stream=True) as stream: + for chunk in stream: + acc.observe(chunk.model_dump(warnings=False)) + message = acc.finalize()["choices"][0]["message"] + reasoning = message["reasoning_content"] or "" + elif surface == "nonstream": + message = client.chat.completions.create(**kwargs).choices[0].message + reasoning = extract_reasoning(agent, message) or "" + else: + if surface == "aux-sync": + stream = client.chat.completions.create(**kwargs, stream=True) + response = _aggregate_chat_stream(stream, model="probe") + else: + response = asyncio.run(aux_async()) + message = response.choices[0].message + reasoning = message.reasoning or "" + content = message["content"] if isinstance(message, dict) else message.content + row.update(reasoning=reasoning, content=content) + row["ok"] = reasoning == expected and content == "LOCAL_CAPTURE_OK" + if surface == "main": + row["ok"] &= all(isinstance(value, str) for value in callbacks) + if row["ok"]: + row["ok"] &= "".join(callbacks) == expected + except Exception as error: + row.update(ok=False, error=f"{type(error).__name__}: {error}") + rows.append(row) + finally: + client.close() + agent.client.close() + server.shutdown() + server.server_close() + thread.join() + return {"fidelity": "Local HTTP fixture + real SDK + production consumers; not vendor inference", + "module": str(Path(__file__).resolve()), "requests": requests, "results": rows} + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--output", required=True) + args = parser.parse_args() + for key in list(os.environ): + if any(token in key.upper() for token in ("TOKEN", "SECRET", "API_KEY", "AUTH")): + os.environ.pop(key, None) + with tempfile.TemporaryDirectory(prefix="reasoning-wire-") as home: + os.environ.update(HOME=home, HERMES_HOME=home) + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + result = run_matrix(("main", "relay", "aux-sync", "aux-async", "nonstream")) + Path(args.output).write_text(json.dumps(result, indent=2), encoding="utf-8") + print(json.dumps({"cases": len(result["results"]), + "passed": sum(row["ok"] for row in result["results"]), + "output": args.output})) + + +if __name__ == "__main__": + main() diff --git a/tests/agent/test_reasoning_shape_boundaries.py b/tests/agent/test_reasoning_shape_boundaries.py new file mode 100644 index 0000000000..cd842b3321 --- /dev/null +++ b/tests/agent/test_reasoning_shape_boundaries.py @@ -0,0 +1,14 @@ +"""Structured reasoning keeps text and answer channels intact on every chat consumer.""" +from evals.providers.reasoning_shapes import run_matrix + + +def test_stream_reasoning_shapes_preserve_text_and_answer(): + result = run_matrix(("main", "relay", "aux-sync", "aux-async")) + failures = [row for row in result["results"] if not row["ok"]] + assert not failures, failures + + +def test_completed_reasoning_shapes_preserve_text_and_answer(): + result = run_matrix(("nonstream",)) + failures = [row for row in result["results"] if not row["ok"]] + assert not failures, failures diff --git a/website/docs/developer-guide/provider-runtime.md b/website/docs/developer-guide/provider-runtime.md index 3e2c723a6b..5cb620a6d1 100644 --- a/website/docs/developer-guide/provider-runtime.md +++ b/website/docs/developer-guide/provider-runtime.md @@ -27,6 +27,16 @@ Primary implementation: If you are trying to add a new first-class inference provider, read [Adding Providers](./adding-providers.md) and the [Model Provider Plugin guide](./model-provider-plugin.md) alongside this page. +## Chat-completions reasoning shapes + +OpenAI-compatible relays can return `reasoning` or `reasoning_content` as strings, +text-part dictionaries, or lists of text parts and string fragments. Hermes flattens +these fields before string operations in the main stream, Relay recording, synchronous +and asynchronous auxiliary streams, and completed-response reasoning extraction. +Fragments retain their explicit whitespace; normalization adds no intra-field separator. +Main-stream and Relay recording retain the existing paragraph breaks between complete +bold reasoning headings. Reasoning stays separate from the visible answer. + ## Resolution precedence At a high level, provider resolution uses: