From 97a68ffb053b3fdc58cccd405934213862420a5b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=81=E5=B1=BF?= Date: Thu, 22 Oct 2026 11:47:58 +0800 Subject: [PATCH] fix(strands): normalize reasoning and agent response boundaries --- .../CHANGELOG.md | 4 + .../README.md | 7 +- .../instrumentation/strands/_hooks.py | 24 ++- .../tests/test_instrumentor.py | 201 ++++++++++++++++++ 4 files changed, 234 insertions(+), 2 deletions(-) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md index f382ff34d..7aee3f080 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Normalize public Strands reasoning text as GenAI reasoning parts on model + spans and conversation history, excluding signatures and redacted content. +- Keep agent output limited to the final response while retaining reasoning + and tool transcripts on the corresponding model and tool spans. - Tolerate the framework-owned baggage context detach that runs when an application closes a strand stream from another context, so `strands-agents` 1.60.0 and later no longer surface a spurious `Failed to detach context` diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md b/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md index c77347092..ede042dd0 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md @@ -55,6 +55,12 @@ invoke_agent (INTERNAL, gen_ai.span.kind=AGENT) suppress model-provider instrumentation. - Standard LoongSuite content-capture settings control prompt and response attributes; content is not captured by default. +- With content capture enabled, model spans and model-input history represent + public reasoning text as `type=reasoning` parts. Reasoning signatures and + encrypted or redacted reasoning are omitted from telemetry. +- Agent output contains the final response, excluding intermediate reasoning, + tool plans, and tool results. Model and tool spans retain their own content; + the application's Strands messages and results are unchanged. - Stream cleanup failures are isolated from application control flow. If an invocation closes before Strands emits its normal after-events, unfinished spans use a synthetic `RuntimeError` describing that incomplete lifecycle. @@ -70,4 +76,3 @@ invoke_agent (INTERNAL, gen_ai.span.kind=AGENT) instrumentation tolerates that framework-owned detach and keeps its own context tokens strictly paired, so closing a stream does not surface a spurious `Failed to detach context` error. - diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_hooks.py b/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_hooks.py index fb0d31c04..093808fa8 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_hooks.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_hooks.py @@ -44,6 +44,7 @@ InputMessage, LLMInvocation, OutputMessage, + Reasoning, Text, ToolCall, ToolCallResponse, @@ -270,7 +271,7 @@ def _after_invocation(self, event: AfterInvocationEvent) -> None: if result is not None: reason = _finish_reason(result.stop_reason) state.agent.finish_reasons = [reason] - state.agent.output_messages = _convert_output_message( + state.agent.output_messages = _convert_agent_output_message( result.message, reason ) _apply_usage( @@ -645,6 +646,19 @@ def _convert_output_message(message: Any, reason: str) -> list[OutputMessage]: ] +def _convert_agent_output_message( + message: Any, reason: str +) -> list[OutputMessage]: + messages = _convert_output_message(message, reason) + for output in messages: + output.parts = [ + part + for part in output.parts + if not isinstance(part, (Reasoning, ToolCall, ToolCallResponse)) + ] + return [output for output in messages if output.parts] + + def _convert_content(blocks: Any) -> list[Any]: if isinstance(blocks, str): return [Text(content=blocks)] @@ -654,6 +668,14 @@ def _convert_content(blocks: Any) -> list[Any]: parts.append(Text(content=block)) elif isinstance(block, dict) and "text" in block: parts.append(Text(content=block["text"])) + elif isinstance(block, dict) and "reasoningContent" in block: + reasoning = block["reasoningContent"] + if isinstance(reasoning, dict): + reasoning_text = reasoning.get("reasoningText") + if isinstance(reasoning_text, dict): + text = reasoning_text.get("text") + if isinstance(text, str) and text: + parts.append(Reasoning(content=text)) elif isinstance(block, dict) and "toolUse" in block: tool_use = block["toolUse"] parts.append( diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py b/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py index f2c0bdbf7..61a376023 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py @@ -20,6 +20,7 @@ import json import os from collections.abc import AsyncGenerator, Sequence +from dataclasses import asdict from types import SimpleNamespace from typing import Any @@ -97,6 +98,25 @@ async def stream( "contentBlockDelta": {"delta": {"text": content["text"]}} } yield {"contentBlockStop": {}} + elif "reasoningContent" in content: + reasoning = content["reasoningContent"]["reasoningText"] + yield {"contentBlockStart": {"start": {}}} + text = reasoning["text"] + for fragment in (text[:7], text[7:]): + yield { + "contentBlockDelta": { + "delta": {"reasoningContent": {"text": fragment}} + } + } + if signature := reasoning.get("signature"): + yield { + "contentBlockDelta": { + "delta": { + "reasoningContent": {"signature": signature} + } + } + } + yield {"contentBlockStop": {}} elif "toolUse" in content: stop_reason = "tool_use" tool_use = content["toolUse"] @@ -421,6 +441,187 @@ async def test_capture_content_includes_every_model_round(content_telemetry): ] == [{"text": "4"}] +@pytest.mark.asyncio +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("capture", ["SPAN_ONLY", "NO_CONTENT"]) +async def test_reasoning_model_rounds_and_final_agent_boundary( + content_telemetry, monkeypatch, streaming, capture +): + monkeypatch.setenv( + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", capture + ) + _, exporter, _ = content_telemetry + + async def worker(label): + responses = [] + for turn in (1, 2): + responses.extend( + [ + { + "content": [ + { + "reasoningContent": { + "reasoningText": { + "text": f"{label} plan {turn}", + "signature": "synthetic-signature", + } + } + }, + { + "toolUse": { + "toolUseId": f"{label}-tool-{turn}", + "name": "calculator", + "input": {"expression": "2+2"}, + } + }, + ] + }, + { + "content": [ + { + "reasoningContent": { + "reasoningText": { + "text": f"{label} conclude {turn}", + "signature": "synthetic-signature", + } + } + }, + {"text": f"{label} answer {turn}: 4"}, + ] + }, + ] + ) + model = DeterministicModel(responses) + agent = _agent(model=model, name=label, callback_handler=None) + for turn in (1, 2): + if streaming: + events = [ + event + async for event in agent.stream_async( + f"{label} turn {turn}" + ) + ] + terminal = [ + event["result"] for event in events if "result" in event + ] + assert len(terminal) == 1 + result = terminal[0] + assert ( + "".join( + event["data"] + for event in events + if isinstance(event.get("data"), str) + ) + == f"{label} answer {turn}: 4" + ) + else: + result = await agent.invoke_async(f"{label} turn {turn}") + assert str(result).strip() == f"{label} answer {turn}: 4" + # Filtering telemetry must never mutate the SDK's business result. + assert result.message["content"][0]["reasoningContent"][ + "reasoningText" + ] == { + "text": f"{label} conclude {turn}", + "signature": "synthetic-signature", + } + assert model.index == 4 + + await asyncio.gather(worker("reasoning-a"), worker("reasoning-b")) + spans = exporter.get_finished_spans() + llms = [s for s in spans if s.attributes["gen_ai.span.kind"] == "LLM"] + agents = [s for s in spans if s.attributes["gen_ai.span.kind"] == "AGENT"] + tools = [s for s in spans if s.attributes["gen_ai.span.kind"] == "TOOL"] + assert (len(llms), len(agents), len(tools)) == (8, 4, 4) + if capture == "NO_CONTENT": + for span in spans: + assert "gen_ai.input.messages" not in span.attributes + assert "gen_ai.output.messages" not in span.attributes + return + + history_reasoning = [] + for span in llms: + output = json.loads(span.attributes["gen_ai.output.messages"]) + assert output[0]["parts"][0]["type"] == "reasoning" + assert set(output[0]["parts"][0]) == {"type", "content"} + history_reasoning.extend( + part + for message in json.loads(span.attributes["gen_ai.input.messages"]) + for part in message["parts"] + if part.get("type") == "reasoning" + ) + assert ( + "synthetic-signature" + not in span.attributes["gen_ai.output.messages"] + ) + assert history_reasoning + for span in agents: + label = span.attributes["gen_ai.agent.name"] + output = json.loads(span.attributes["gen_ai.output.messages"]) + assert len(output) == 1 + assert len(output[0]["parts"]) == 1 + assert output[0]["parts"][0]["type"] == "text" + assert output[0]["parts"][0]["content"].startswith(f"{label} answer ") + + +@pytest.mark.parametrize( + "reasoning", + [ + {"reasoningText": {"signature": "synthetic-signature"}}, + {"reasoningText": {"text": ""}}, + {"reasoningText": {"text": 42}}, + {"redactedContent": b"synthetic-encrypted"}, + {"reasoningText": "malformed"}, + None, + ], +) +def test_unavailable_reasoning_does_not_export_opaque_content(reasoning): + messages = hooks_module._convert_output_message( + { + "role": "assistant", + "content": [ + {"reasoningContent": reasoning}, + {"text": "Final answer"}, + ], + }, + "stop", + ) + assert asdict(messages[0])["parts"] == [ + {"content": "Final answer", "type": "text"} + ] + + +def test_agent_boundary_excludes_tool_transcript_and_keeps_final_parts(): + final_image = { + "image": { + "format": "png", + "source": {"s3Location": {"uri": "s3://example/final.png"}}, + } + } + message = { + "role": "assistant", + "content": [ + {"reasoningContent": {"reasoningText": {"text": "Thought"}}}, + {"toolUse": {"toolUseId": "tool-1", "name": "calculator"}}, + {"toolResult": {"toolUseId": "tool-1", "content": []}}, + final_image, + {"text": "Final answer"}, + ], + } + output = hooks_module._convert_agent_output_message(message, "stop") + assert asdict(output[0])["parts"] == [ + final_image, + {"content": "Final answer", "type": "text"}, + ] + assert len(message["content"]) == 5 + assert ( + hooks_module._convert_agent_output_message( + {"role": "assistant", "content": message["content"][:3]}, + "tool_calls", + ) + == [] + ) + + @pytest.mark.asyncio async def test_framework_and_provider_llm_spans_can_coexist(telemetry): instrumentor, exporter, _ = telemetry