fix(agentscope): start streaming LLM spans before model calls (#242)

Author: sipercai

instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Fixed
+
+- Start AgentScope v2 streaming LLM spans before invoking the underlying model
+  call so TTFT and stream lifecycle are recorded on the framework LLM span.
+- Capture AgentScope v2 string message content as text parts so LLM input and
+  output message attributes are populated when content capture is enabled.
+
 ## Version 0.7.0 (2026-07-03)
 
 ### Added

instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py

@@ -31,7 +31,7 @@
 from agentscope.model import ChatModelBase, ChatResponse
 from agentscope.tool import ToolResponse
 
-from opentelemetry.context import Context, get_current
+from opentelemetry.context import get_current
 from opentelemetry.util.genai.extended_handler import ExtendedTelemetryHandler
 from opentelemetry.util.genai.extended_types import (
     ExecuteToolInvocation,
@@ -175,30 +175,20 @@ async def on_model_call(
 
         invocation = _create_llm_invocation(model, input_kwargs)
         span_context = get_current()
-        started = False
-        if not _is_streaming_model(model, input_kwargs):
-            handler.start_llm(invocation, context=span_context)
-            started = True
+        handler.start_llm(invocation, context=span_context)
         try:
             result = await next_handler(**input_kwargs)
             if inspect.isasyncgen(result):
                 return self._wrap_model_stream(
                     result,
                     invocation,
-                    span_context,
                     handler,
-                    span_started=started,
                 )
 
-            if not started:
-                handler.start_llm(invocation, context=span_context)
-                started = True
             _finish_llm_invocation(invocation, result)
             handler.stop_llm(invocation)
             return result
         except BaseException as exc:
-            if not started:
-                handler.start_llm(invocation, context=span_context)
             handler.fail_llm(
                 invocation,
                 Error(message=str(exc) or type(exc).__name__, type=type(exc)),
@@ -209,17 +199,11 @@ async def _wrap_model_stream(
         self,
         result: AsyncGenerator[ChatResponse, None],
         invocation: LLMInvocation,
-        span_context: Context,
         handler: ExtendedTelemetryHandler,
-        *,
-        span_started: bool,
     ) -> AsyncGenerator[ChatResponse, None]:
         first_token_seen = False
         last_chunk = None
         closed = False
-        if not span_started:
-            handler.start_llm(invocation, context=span_context)
-            span_started = True
         try:
             async for chunk in result:
                 if not first_token_seen:
@@ -239,7 +223,7 @@ async def _wrap_model_stream(
             handler.stop_llm(invocation)
             closed = True
         finally:
-            if span_started and not closed:
+            if not closed:
                 handler.stop_llm(invocation)
 
     async def on_acting(
@@ -402,6 +386,10 @@ def _chat_response_to_output(response: ChatResponse) -> OutputMessage:
 
 
 def _blocks_to_parts(blocks: Sequence[Any]) -> list[Any]:
+    if blocks is None:
+        return []
+    if isinstance(blocks, str):
+        return [Text(content=blocks)]
     parts = []
     for block in blocks:
         block_type = getattr(block, "type", None)

instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_v2_instrumentation.py

@@ -18,6 +18,7 @@
 
 import asyncio
 import importlib.metadata
+import json
 import os
 from dataclasses import asdict
 from types import SimpleNamespace
@@ -41,13 +42,17 @@
 from agentscope.model import ChatResponse, DashScopeChatModel  # noqa: E402
 from agentscope.tool import ToolResponse  # noqa: E402
 
+from opentelemetry import trace as trace_api  # noqa: E402
 from opentelemetry.instrumentation.agentscope._v2_middleware import (  # noqa: E402
     AgentScopeV2Middleware,
     _message_to_input,
 )
 from opentelemetry.instrumentation.agentscope.package import (  # noqa: E402
     get_installed_instrumentation_dependencies,
 )
+from opentelemetry.semconv._incubating.attributes import (  # noqa: E402
+    gen_ai_attributes as GenAI,
+)
 from opentelemetry.trace.status import StatusCode  # noqa: E402
 from opentelemetry.util.genai.utils import gen_ai_json_dumps  # noqa: E402
 
@@ -209,6 +214,114 @@ async def stream_handler(**kwargs):
     assert span.attributes["error.type"] == "RuntimeError"
 
 
+async def test_v2_streaming_model_call_starts_llm_span_before_model_handler(
+    instrument,
+    span_exporter,
+):
+    agent = Agent(
+        name="stream_suppression_agent",
+        system_prompt="Reply briefly.",
+        model=_make_model(stream=True),
+    )
+    middleware = _middleware(agent._model_call_middlewares)
+    observed_current_span_ids = []
+    consumer_current_span_ids = []
+
+    async def stream_handler(**kwargs):
+        del kwargs
+        observed_current_span_ids.append(
+            trace_api.get_current_span().get_span_context().span_id
+        )
+
+        async def stream():
+            observed_current_span_ids.append(
+                trace_api.get_current_span().get_span_context().span_id
+            )
+            yield ChatResponse(
+                content=[TextBlock(text="partial")],
+                is_last=False,
+            )
+            observed_current_span_ids.append(
+                trace_api.get_current_span().get_span_context().span_id
+            )
+            yield ChatResponse(
+                content=[TextBlock(text="done")],
+                is_last=True,
+            )
+
+        return stream()
+
+    stream = await middleware.on_model_call(
+        agent,
+        {
+            "current_model": agent.model,
+            "messages": [UserMsg(name="user", content="hello")],
+        },
+        stream_handler,
+    )
+
+    async for _ in stream:
+        consumer_current_span_ids.append(
+            trace_api.get_current_span().get_span_context().span_id
+        )
+
+    spans = _spans_by_operation(span_exporter.get_finished_spans(), "chat")
+    assert len(spans) == 1
+    llm_span_id = spans[0].context.span_id
+    assert observed_current_span_ids == [llm_span_id, llm_span_id, llm_span_id]
+    assert consumer_current_span_ids == [llm_span_id, llm_span_id]
+    assert (
+        trace_api.get_current_span().get_span_context().span_id != llm_span_id
+    )
+
+
+async def test_v2_streaming_model_call_captures_input_and_output_content(
+    instrument_with_content,
+    span_exporter,
+):
+    agent = Agent(
+        name="stream_content_agent",
+        system_prompt="Reply briefly.",
+        model=_make_model(stream=True),
+    )
+    middleware = _middleware(agent._model_call_middlewares)
+
+    async def stream_handler(**kwargs):
+        del kwargs
+
+        async def stream():
+            yield ChatResponse(
+                content=[TextBlock(text="partial")],
+                is_last=False,
+            )
+            yield ChatResponse(
+                content=[TextBlock(text="done")],
+                is_last=True,
+            )
+
+        return stream()
+
+    stream = await middleware.on_model_call(
+        agent,
+        {
+            "current_model": agent.model,
+            "messages": [UserMsg(name="user", content="hello")],
+        },
+        stream_handler,
+    )
+
+    async for _ in stream:
+        pass
+
+    span = _spans_by_operation(span_exporter.get_finished_spans(), "chat")[0]
+    input_messages = json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES])
+    output_messages = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES])
+    assert input_messages[0]["role"] == "user"
+    assert input_messages[0]["parts"][0]["content"] == "hello"
+    assert output_messages[0]["role"] == "assistant"
+    assert output_messages[0]["parts"][0]["content"] == "done"
+
+
 async def test_v2_tool_acting_hook(instrument, span_exporter):
     agent = Agent(
         name="tool_agent",