[feature request] Support gen_ai.system.message events in strands-agents processor

[AmirSaeed96]

Add support for system messages in openinference-instrumentation-strands-agents

Description

The openinference-instrumentation-strands-agents processor currently transforms user, assistant, tool, and choice messages from GenAI events to OpenInference format, but does not process system messages. This means system prompts configured in Strands agents are not included in the transformed spans.

Adding system message support would enable users to see system prompts in their observability tools (Phoenix, Arize AX) and use them in evaluators.

Current Behavior

When a Strands agent is configured with a system prompt:

from strands import Agent
from strands.models.openai import OpenAIModel

agent = Agent(
    name="WeatherAssistant",
    model=OpenAIModel(model_id="gpt-4o-mini"),
    system_prompt="You are a helpful weather assistant."
)

The native Strands instrumentation emits gen_ai.system.message events, but the OpenInference processor does not extract them. The transformed span contains only user and assistant messages:

{
  "llm.input_messages": [
    {
      "message.role": "user",
      "message.content": "What's the weather?"
    }
  ]
}

Expected Behavior

System messages should be extracted from gen_ai.system.message events and prepended to llm.input_messages:

{
  "llm.input_messages": [
    {
      "message.role": "system",
      "message.content": "You are a helpful weather assistant."
    },
    {
      "message.role": "user",
      "message.content": "What's the weather?"
    }
  ]
}

This would align with OpenAI's chat completion format and make system prompts visible in the observability UI.

Implementation Reference

The openinference-instrumentation-pydantic-ai package already implements this pattern correctly:

File: python/instrumentation/openinference-instrumentation-pydantic-ai/src/openinference/instrumentation/pydantic_ai/semantic_conventions.py

if event.get(OTELConventions.EVENT_NAME) == GenAIEventNames.SYSTEM_MESSAGE:
    message = {}
    if GenAIMessageFields.ROLE in event:
        message[MessageAttributes.MESSAGE_ROLE] = event[GenAIMessageFields.ROLE]
    if GenAIMessageFields.CONTENT in event:
        message[MessageAttributes.MESSAGE_CONTENT] = event[GenAIMessageFields.CONTENT]
    if message:
        input_messages.append(message)

It also handles the gen_ai.system_instructions attribute:

if GEN_AI_SYSTEM_INSTRUCTIONS in gen_ai_attrs:
    system_instructions = json.loads(gen_ai_attrs[GEN_AI_SYSTEM_INSTRUCTIONS])
    if isinstance(system_instructions, list):
        for system_instruction in system_instructions:
            yield (
                f"{SpanAttributes.LLM_INPUT_MESSAGES}.{msg_index}.{MessageAttributes.MESSAGE_ROLE}",
                GenAIMessageRoles.SYSTEM,
            )
            yield (
                f"{SpanAttributes.LLM_INPUT_MESSAGES}.{msg_index}.{MessageAttributes.MESSAGE_CONTENT}",
                system_instruction[GenAISystemInstructionsFields.CONTENT],
            )

Suggested Implementation

File: python/instrumentation/openinference-instrumentation-strands-agents/src/openinference/instrumentation/strands_agents/processor.py

1. Add system message event handling in _extract_messages_from_events()

def _extract_messages_from_events(
    self, events: List[Any]
) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    """Extract input and output messages from Strands events."""
    input_messages = []
    output_messages = []

    for event in events:
        event_name = (
            getattr(event, "name", "") if hasattr(event, "name") else event.get("name", "")
        )
        event_attrs = (
            getattr(event, "attributes", {})
            if hasattr(event, "attributes")
            else event.get("attributes", {})
        )

        # Add this block:
        if event_name == GenAIEventNames.SYSTEM_MESSAGE:
            content = event_attrs.get("content", "")
            message = self._parse_message_content(content, "system")
            if message:
                # System messages should appear first
                input_messages.insert(0, message)

        elif event_name == GenAIEventNames.USER_MESSAGE:
            # ... existing code ...

2. Add gen_ai.system_instructions attribute handling

In the _transform_attributes() method, after message extraction from events:

# Handle gen_ai.system_instructions attribute if present
system_instructions_attr = attrs.get("gen_ai.system_instructions")
if system_instructions_attr:
    system_message = self._extract_system_instructions(system_instructions_attr)
    if system_message:
        input_messages.insert(0, system_message)

3. Add helper method for parsing system instructions

def _extract_system_instructions(self, system_instructions: Any) -> Optional[Dict[str, Any]]:
    """
    Extract system instructions from gen_ai.system_instructions attribute.

    Supports both plain string format and structured JSON format.
    """
    if not system_instructions:
        return None

    try:
        if isinstance(system_instructions, str):
            try:
                parsed = json.loads(system_instructions)
            except json.JSONDecodeError:
                return {"message.role": "system", "message.content": system_instructions}
        else:
            parsed = system_instructions

        if isinstance(parsed, list):
            text_parts = []
            for instruction in parsed:
                if isinstance(instruction, dict) and "content" in instruction:
                    text_parts.append(str(instruction["content"]))
                elif isinstance(instruction, str):
                    text_parts.append(instruction)
            if text_parts:
                return {"message.role": "system", "message.content": " ".join(text_parts)}

        elif isinstance(parsed, dict) and "content" in parsed:
            return {"message.role": "system", "message.content": str(parsed["content"])}

    except Exception as e:
        logger.warning(f"Failed to extract system instructions: {e}")

    return None

Testing

Test cases should cover:

1. System message event extraction
2. System instructions attribute extraction
3. Message ordering (system messages should be first)
4. Multiple system messages (edge case)
5. Both plain string and JSON formats

Example test:

def test_system_message_event_extraction():
    processor = StrandsAgentsToOpenInferenceProcessor()

    events = [
        {
            "name": "gen_ai.system.message",
            "attributes": {"content": "You are a helpful assistant."}
        },
        {
            "name": "gen_ai.user.message",
            "attributes": {"content": "Hello!"}
        }
    ]

    input_messages, _ = processor._extract_messages_from_events(events)

    assert len(input_messages) == 2
    assert input_messages[0]["message.role"] == "system"
    assert input_messages[0]["message.content"] == "You are a helpful assistant."
    assert input_messages[1]["message.role"] == "user"

Benefits

• Complete context for evaluators: System prompts provide important context about agent behavior, constraints, and persona that evaluators need to assess output quality
• Consistency: Aligns with how other OpenInference instrumentation packages (pydantic-ai, etc.) handle system messages
• Standards compliance: Matches OpenTelemetry GenAI semantic conventions and OpenAI chat format
• User experience: Users can see the full conversation context in observability UIs

Additional Context

The GenAIEventNames.SYSTEM_MESSAGE constant is already defined in semantic_conventions.py but not currently used by the processor.

Environment

• Package: openinference-instrumentation-strands-agents
• Affected versions: 0.1.0 through 0.1.8 (current latest)
• Python version: 3.9+