Let OTel events from Logfire be deserialized into `ModelMessage`s

[DouweM]

This would be the inverse of InstrumentationSettings.messages_to_otel_events.

This would allow re-running a past conversation by passing these messages to agent.run as message_history.

I believe all the necessary information to build the messages and their parts are in the events.

[adtyavrdhn]

What about when content is not included?

Example:

If it has been instrumented with:
InstrumentationSettings(include_content=False)

Message -> ModelRequest(parts=[UserPromptPart('simple text prompt')])

Event ->

{
    'content': {'kind': 'text'},
    'role': 'user',
    'gen_ai.message.index': 7,
    'event.name': 'gen_ai.user.message',
}

We would end up with something like

ModelRequest(parts=[UserPromptPart(content="")])

[DouweM]

Here's a script @alexmojaki wrote at some point to do this:

import itertools
import json
from typing import Any

import logfire
import pydantic

from pydantic_ai.direct import model_request_sync
from pydantic_ai.messages import (
    ModelMessage,
    ModelRequest,
    ModelResponse,
    SystemPromptPart,
    TextPart,
    ToolCallPart,
    ToolReturnPart,
    UserPromptPart,
)
from pydantic_ai.models import ModelRequestParameters

logfire.configure()
logfire.instrument_pydantic_ai()

attributes = json.loads(r"""
{ "events": [ { "content": "Be concise, reply with one sentence.Use the `get_lat_lng` tool to get the latitude and longitude of the locations, then use the `get_weather` tool to get the weather.", "role": "system", "gen_ai.system": "openai", "event.name": "gen_ai.system.message" }, { "content": "What is the weather like in London and in Wiltshire?", "role": "user", "gen_ai.system": "openai", "gen_ai.message.index": 0, "event.name": "gen_ai.user.message" }, { "role": "assistant", "tool_calls": [ { "id": "call_X4FLlIQ4l3YWQzuYzAkS1ibA", "type": "function", "function": { "name": "get_lat_lng", "arguments": "{\"location_description\": \"London, UK\"}" } }, { "id": "call_MSI9EjdOJvmkjRCZPEPtbjda", "type": "function", "function": { "name": "get_lat_lng", "arguments": "{\"location_description\": \"Wiltshire, UK\"}" } } ], "gen_ai.system": "openai", "gen_ai.message.index": 1, "event.name": "gen_ai.assistant.message" }, { "content": { "lat": 51.1, "lng": -0.1 }, "role": "tool", "id": "call_X4FLlIQ4l3YWQzuYzAkS1ibA", "name": "get_lat_lng", "gen_ai.system": "openai", "gen_ai.message.index": 2, "event.name": "gen_ai.tool.message" }, { "content": { "lat": 51.1, "lng": -0.1 }, "role": "tool", "id": "call_MSI9EjdOJvmkjRCZPEPtbjda", "name": "get_lat_lng", "gen_ai.system": "openai", "gen_ai.message.index": 2, "event.name": "gen_ai.tool.message" }, { "role": "assistant", "tool_calls": [ { "id": "call_qhLCZltqVWYiNTgUGSeNRa3K", "type": "function", "function": { "name": "get_weather", "arguments": "{\"lat\": 51.1, \"lng\": -0.1}" } }, { "id": "call_vCwGKWZurQAMN8hk2WO9M8U0", "type": "function", "function": { "name": "get_weather", "arguments": "{\"lat\": 51.1, \"lng\": -0.1}" } } ], "gen_ai.system": "openai", "gen_ai.message.index": 3, "event.name": "gen_ai.assistant.message" }, { "content": { "temperature": "21 °C", "description": "Sunny" }, "role": "tool", "id": "call_qhLCZltqVWYiNTgUGSeNRa3K", "name": "get_weather", "gen_ai.system": "openai", "gen_ai.message.index": 4, "event.name": "gen_ai.tool.message" }, { "content": { "temperature": "21 °C", "description": "Sunny" }, "role": "tool", "id": "call_vCwGKWZurQAMN8hk2WO9M8U0", "name": "get_weather", "gen_ai.system": "openai", "gen_ai.message.index": 4, "event.name": "gen_ai.tool.message" }, { "index": 0, "message": { "role": "assistant", "content": "The weather in both London and Wiltshire is currently 21 °C and sunny." }, "gen_ai.system": "openai", "event.name": "gen_ai.choice" } ], "gen_ai.operation.name": "chat", "gen_ai.request.model": "gpt-4o", "gen_ai.response.model": "gpt-4o-2024-08-06", "gen_ai.system": "openai", "gen_ai.usage.input_tokens": 351, "gen_ai.usage.output_tokens": 18, "model_request_parameters": { "function_tools": [ { "name": "get_lat_lng", "description": "Get the latitude and longitude of a location.", "parameters_json_schema": { "additionalProperties": false, "properties": { "location_description": { "description": "A description of a location.", "type": "string" } }, "required": [ "location_description" ], "type": "object" }, "outer_typed_dict_key": null, "strict": true }, { "name": "get_weather", "description": "Get the weather at a location.", "parameters_json_schema": { "additionalProperties": false, "properties": { "lat": { "description": "Latitude of the location.", "type": "number" }, "lng": { "description": "Longitude of the location.", "type": "number" } }, "required": [ "lat", "lng" ], "type": "object" }, "outer_typed_dict_key": null, "strict": true } ], "allow_text_output": true, "output_tools": [] }, "server.address": "api.openai.com" }
""")

ModelRequestParametersAdapter = pydantic.TypeAdapter(ModelRequestParameters)

events = attributes['events']
messages: list[ModelMessage] = []

for _, event_group in itertools.groupby(events, key=lambda e: e.get('gen_ai.message.index')):
    event_group = list(event_group)
    if event_group[0]['event.name'] == 'gen_ai.choice':
        break
    parts: list[Any] = []
    message_cls = None
    for event in event_group:
        role = event['role']
        current_message_cls = ModelRequest
        if role == 'user':
            parts.append(UserPromptPart(event['content']))
        elif role == 'system':
            parts.append(SystemPromptPart(event['content']))
        elif role == 'tool':
            parts.append(ToolReturnPart(tool_name=event['name'], tool_call_id=event['id'], content=event['content']))
        else:
            assert role == 'assistant'
            if 'content' in event:
                parts.append(TextPart(event['content']))
            for tool_call in event.get('tool_calls', []):
                parts.append(
                    ToolCallPart(
                        tool_call_id=tool_call['id'],
                        tool_name=tool_call['function']['name'],
                        args=tool_call['function']['arguments'],
                    )
                )
            current_message_cls = ModelResponse
        assert message_cls in (None, current_message_cls)
        message_cls = current_message_cls
    assert message_cls
    message = message_cls(parts)
    messages.append(message)

model_request_parameters = ModelRequestParametersAdapter.validate_python(attributes['model_request_parameters'])
print(messages)
print(model_request_parameters)
model_request_sync(
    attributes['gen_ai.response.model'],
    messages,
    model_request_parameters=model_request_parameters,
)

It's missing ThinkingPart and possibly some other newer additions, but would be a good starting point.

[DouweM]

@adtyavrdhn Yep this wouldn't work if content is excluded

[adtyavrdhn]

I'd like to work on this.

[alexmojaki]

I think work on this should only come after #2349 is finished

[adtyavrdhn]

Okay yeah, that makes sense. I'll take it up once you're done with #2349