diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md index ecf0e4fdb..da5f3c297 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md +++ b/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 diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py index d631691e4..982adf987 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py +++ b/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) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_v2_instrumentation.py b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_v2_instrumentation.py index b4f705ba6..8b73b572a 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_v2_instrumentation.py +++ b/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,6 +42,7 @@ 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, @@ -48,6 +50,9 @@ 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",