wehub-resource-sync
9201ef759e
Harness Compat / harness compat (push) Failing after 0s
CI / test on 3.12 (standard) (push) Has been cancelled
CI / test on 3.13 (standard) (push) Has been cancelled
CI / test on 3.14 (standard) (push) Has been cancelled
CI / test on 3.10 (all-extras) (push) Has been cancelled
CI / test on 3.11 (all-extras) (push) Has been cancelled
CI / test on 3.12 (all-extras) (push) Has been cancelled
CI / test on 3.14 (pydantic-ai-slim) (push) Has been cancelled
CI / test on 3.10 (pydantic-evals) (push) Has been cancelled
CI / test on 3.11 (pydantic-evals) (push) Has been cancelled
CI / test on 3.12 (pydantic-evals) (push) Has been cancelled
CI / deploy-docs-preview (push) Has been cancelled
CI / build release artifacts (push) Has been cancelled
CI / publish to PyPI (push) Has been cancelled
CI / Send tweet (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / mypy (push) Has been cancelled
CI / docs (push) Has been cancelled
CI / test on 3.10 (standard) (push) Has been cancelled
CI / test on 3.11 (standard) (push) Has been cancelled
CI / test on 3.13 (all-extras) (push) Has been cancelled
CI / test on 3.14 (all-extras) (push) Has been cancelled
CI / test on 3.10 (pydantic-ai-slim) (push) Has been cancelled
CI / test on 3.11 (pydantic-ai-slim) (push) Has been cancelled
CI / test on 3.12 (pydantic-ai-slim) (push) Has been cancelled
CI / test on 3.13 (pydantic-ai-slim) (push) Has been cancelled
CI / test on 3.13 (pydantic-evals) (push) Has been cancelled
CI / test on 3.14 (pydantic-evals) (push) Has been cancelled
CI / test on 3.10 (lowest-versions) (push) Has been cancelled
CI / test on 3.11 (lowest-versions) (push) Has been cancelled
CI / test on 3.12 (lowest-versions) (push) Has been cancelled
CI / test on 3.13 (lowest-versions) (push) Has been cancelled
CI / test on 3.14 (lowest-versions) (push) Has been cancelled
CI / test examples on 3.11 (push) Has been cancelled
CI / test examples on 3.12 (push) Has been cancelled
CI / test examples on 3.13 (push) Has been cancelled
CI / test examples on 3.14 (push) Has been cancelled
CI / coverage (push) Has been cancelled
CI / check (push) Has been cancelled
CI / deploy-docs (push) Has been cancelled
2147 lines
82 KiB
Python
2147 lines
82 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncGenerator, AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
|
|
import pytest
|
|
from opentelemetry.trace import NoOpTracerProvider
|
|
|
|
from pydantic_ai import (
|
|
AudioUrl,
|
|
BinaryContent,
|
|
CachePoint,
|
|
DocumentUrl,
|
|
FilePart,
|
|
FinalResultEvent,
|
|
ImageUrl,
|
|
ModelMessage,
|
|
ModelRequest,
|
|
ModelResponse,
|
|
ModelResponseStreamEvent,
|
|
NativeToolCallPart,
|
|
NativeToolReturnPart,
|
|
PartDeltaEvent,
|
|
PartEndEvent,
|
|
PartStartEvent,
|
|
RetryPromptPart,
|
|
SystemPromptPart,
|
|
TextPart,
|
|
TextPartDelta,
|
|
ThinkingPart,
|
|
ToolCallPart,
|
|
ToolReturnPart,
|
|
UserPromptPart,
|
|
VideoUrl,
|
|
)
|
|
from pydantic_ai._run_context import RunContext
|
|
from pydantic_ai._warnings import PydanticAIDeprecationWarning
|
|
from pydantic_ai.models import Model, ModelRequestParameters, StreamedResponse
|
|
from pydantic_ai.models.instrumented import InstrumentationSettings, InstrumentedModel
|
|
from pydantic_ai.settings import ModelSettings
|
|
from pydantic_ai.tools import ToolDefinition
|
|
from pydantic_ai.usage import RequestUsage
|
|
|
|
from .._inline_snapshot import snapshot, warns
|
|
from ..conftest import IsDatetime, IsFloat, IsInt, IsStr, try_import
|
|
|
|
with try_import() as imports_successful:
|
|
from logfire.testing import CaptureLogfire
|
|
|
|
pytestmark = [
|
|
pytest.mark.skipif(not imports_successful(), reason='logfire not installed'),
|
|
pytest.mark.anyio,
|
|
]
|
|
|
|
|
|
def deprecated_instrumentation_settings(
|
|
version: Literal[2, 3, 4], *, include_binary_content: bool = True, include_content: bool = True
|
|
) -> InstrumentationSettings:
|
|
with pytest.warns(
|
|
PydanticAIDeprecationWarning,
|
|
match=r'Instrumentation format versions 2, 3, and 4 are deprecated',
|
|
):
|
|
return InstrumentationSettings(
|
|
version=version, include_binary_content=include_binary_content, include_content=include_content
|
|
)
|
|
|
|
|
|
class MyModel(Model):
|
|
# Use a system and model name that have a known price
|
|
@property
|
|
def system(self) -> str:
|
|
return 'openai'
|
|
|
|
@property
|
|
def model_name(self) -> str:
|
|
return 'gpt-4o'
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return 'https://example.com:8000/foo'
|
|
|
|
async def request(
|
|
self,
|
|
messages: list[ModelMessage],
|
|
model_settings: ModelSettings | None,
|
|
model_request_parameters: ModelRequestParameters,
|
|
) -> ModelResponse:
|
|
return ModelResponse(
|
|
parts=[
|
|
TextPart('text1'),
|
|
ToolCallPart('tool1', 'args1', 'tool_call_1'),
|
|
ToolCallPart('tool2', {'args2': 3}, 'tool_call_2'),
|
|
TextPart('text2'),
|
|
{}, # test unexpected parts # type: ignore
|
|
],
|
|
usage=RequestUsage(
|
|
input_tokens=100,
|
|
output_tokens=200,
|
|
cache_write_tokens=10,
|
|
cache_read_tokens=20,
|
|
input_audio_tokens=10,
|
|
cache_audio_read_tokens=5,
|
|
output_audio_tokens=30,
|
|
details={'reasoning_tokens': 30},
|
|
),
|
|
model_name='gpt-4o-2024-11-20',
|
|
provider_details=dict(finish_reason='stop', foo='bar'),
|
|
provider_response_id='response_id',
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def request_stream(
|
|
self,
|
|
messages: list[ModelMessage],
|
|
model_settings: ModelSettings | None,
|
|
model_request_parameters: ModelRequestParameters,
|
|
run_context: RunContext | None = None,
|
|
) -> AsyncGenerator[StreamedResponse]:
|
|
yield MyResponseStream(model_request_parameters=model_request_parameters)
|
|
|
|
async def count_tokens(
|
|
self,
|
|
messages: list[ModelMessage],
|
|
model_settings: ModelSettings | None,
|
|
model_request_parameters: ModelRequestParameters,
|
|
) -> RequestUsage:
|
|
return RequestUsage(input_tokens=10)
|
|
|
|
|
|
class MyResponseStream(StreamedResponse):
|
|
async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
|
|
self._usage = RequestUsage(input_tokens=300, output_tokens=400)
|
|
for event in self._parts_manager.handle_text_delta(vendor_part_id=0, content='text1'):
|
|
yield event
|
|
for event in self._parts_manager.handle_text_delta(vendor_part_id=0, content='text2'):
|
|
yield event
|
|
|
|
@property
|
|
def model_name(self) -> str:
|
|
return 'gpt-4o-2024-11-20'
|
|
|
|
@property
|
|
def provider_name(self) -> str:
|
|
return 'openai'
|
|
|
|
@property
|
|
def provider_url(self) -> str:
|
|
return 'https://api.openai.com'
|
|
|
|
@property
|
|
def timestamp(self) -> datetime:
|
|
return datetime(2022, 1, 1)
|
|
|
|
|
|
async def test_instrumented_model(capfire: CaptureLogfire):
|
|
model = InstrumentedModel(MyModel(), InstrumentationSettings())
|
|
assert model.system == 'openai'
|
|
assert model.model_name == 'gpt-4o'
|
|
assert model.model_id == 'openai:gpt-4o'
|
|
|
|
messages = [
|
|
ModelRequest(
|
|
instructions='instructions',
|
|
parts=[
|
|
SystemPromptPart('system_prompt'),
|
|
UserPromptPart('user_prompt'),
|
|
ToolReturnPart('tool3', 'tool_return_content', 'tool_call_3'),
|
|
RetryPromptPart('retry_prompt1', tool_name='tool4', tool_call_id='tool_call_4'),
|
|
RetryPromptPart('retry_prompt2'),
|
|
{}, # test unexpected parts # type: ignore
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelResponse(parts=[TextPart('text3')]),
|
|
]
|
|
await model.request(
|
|
messages,
|
|
model_settings=ModelSettings(temperature=1),
|
|
model_request_parameters=ModelRequestParameters(
|
|
function_tools=[],
|
|
allow_text_output=True,
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
),
|
|
)
|
|
|
|
assert capfire.exporter.exported_spans_as_dict(parse_json_attributes=True) == snapshot(
|
|
[
|
|
{
|
|
'name': 'chat gpt-4o',
|
|
'context': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
|
|
'parent': None,
|
|
'start_time': 1000000000,
|
|
'end_time': 2000000000,
|
|
'attributes': {
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'server.address': 'example.com',
|
|
'server.port': 8000,
|
|
'model_request_parameters': {
|
|
'function_tools': [],
|
|
'native_tools': [],
|
|
'output_mode': 'text',
|
|
'output_object': None,
|
|
'output_tools': [],
|
|
'prompted_output_template': None,
|
|
'allow_text_output': True,
|
|
'allow_image_output': False,
|
|
'instruction_parts': None,
|
|
'thinking': None,
|
|
},
|
|
'logfire.json_schema': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'gen_ai.input.messages': {'type': 'array'},
|
|
'gen_ai.output.messages': {'type': 'array'},
|
|
'gen_ai.system_instructions': {'type': 'array'},
|
|
'model_request_parameters': {'type': 'object'},
|
|
},
|
|
},
|
|
'gen_ai.request.temperature': 1,
|
|
'logfire.msg': 'chat gpt-4o',
|
|
'gen_ai.input.messages': [
|
|
{'role': 'system', 'parts': [{'type': 'text', 'content': 'system_prompt'}]},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt'},
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'tool_call_3',
|
|
'name': 'tool3',
|
|
'result': 'tool_return_content',
|
|
},
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'tool_call_4',
|
|
'name': 'tool4',
|
|
'result': """\
|
|
retry_prompt1
|
|
|
|
Fix the errors and try again.\
|
|
""",
|
|
},
|
|
{
|
|
'type': 'text',
|
|
'content': """\
|
|
Validation feedback:
|
|
retry_prompt2
|
|
|
|
Fix the errors and try again.\
|
|
""",
|
|
},
|
|
],
|
|
},
|
|
{'role': 'assistant', 'parts': [{'type': 'text', 'content': 'text3'}]},
|
|
],
|
|
'gen_ai.output.messages': [
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text1'},
|
|
{'type': 'tool_call', 'id': 'tool_call_1', 'name': 'tool1', 'arguments': 'args1'},
|
|
{'type': 'tool_call', 'id': 'tool_call_2', 'name': 'tool2', 'arguments': {'args2': 3}},
|
|
{'type': 'text', 'content': 'text2'},
|
|
],
|
|
}
|
|
],
|
|
'logfire.span_type': 'span',
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
'gen_ai.response.id': 'response_id',
|
|
'gen_ai.system_instructions': [{'type': 'text', 'content': 'instructions'}],
|
|
'gen_ai.usage.cache_creation.input_tokens': 10,
|
|
'gen_ai.usage.cache_read.input_tokens': 20,
|
|
'gen_ai.usage.details.reasoning_tokens': 30,
|
|
'gen_ai.usage.details.cache_write_tokens': 10,
|
|
'gen_ai.usage.details.cache_read_tokens': 20,
|
|
'gen_ai.usage.details.input_audio_tokens': 10,
|
|
'gen_ai.usage.details.cache_audio_read_tokens': 5,
|
|
'gen_ai.usage.details.output_audio_tokens': 30,
|
|
'gen_ai.usage.input_tokens': 100,
|
|
'gen_ai.usage.output_tokens': 200,
|
|
'operation.cost': 0.002225,
|
|
},
|
|
},
|
|
]
|
|
)
|
|
|
|
assert capfire.log_exporter.exported_logs_as_dicts() == snapshot([])
|
|
|
|
|
|
async def test_instrumented_model_not_recording():
|
|
model = InstrumentedModel(
|
|
MyModel(),
|
|
InstrumentationSettings(tracer_provider=NoOpTracerProvider()),
|
|
)
|
|
|
|
messages: list[ModelMessage] = [ModelRequest(parts=[SystemPromptPart('system_prompt')], timestamp=IsDatetime())]
|
|
await model.request(
|
|
messages,
|
|
model_settings=ModelSettings(temperature=1),
|
|
model_request_parameters=ModelRequestParameters(
|
|
function_tools=[],
|
|
allow_text_output=True,
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
),
|
|
)
|
|
|
|
|
|
async def test_instrumented_model_serializes_lone_surrogates_without_crashing(capfire: CaptureLogfire):
|
|
"""Lone surrogates in message content make `to_json` raise; instrumentation must not crash the run.
|
|
|
|
Text decoded with `errors='surrogateescape'` can carry unpaired surrogates. `to_json` rejects
|
|
them, so `handle_messages` falls back to a serializer that escapes them instead of propagating.
|
|
"""
|
|
model = InstrumentedModel(MyModel(), InstrumentationSettings())
|
|
|
|
surrogate = 'before\udce4after'
|
|
messages: list[ModelMessage] = [ModelRequest(parts=[UserPromptPart(surrogate)], timestamp=IsDatetime())]
|
|
await model.request(messages, model_settings=None, model_request_parameters=ModelRequestParameters())
|
|
|
|
attributes = capfire.exporter.exported_spans_as_dict()[0]['attributes']
|
|
assert attributes['gen_ai.input.messages'] == snapshot(
|
|
'[{"role":"user","parts":[{"type":"text","content":"before\\udce4after"}]}]'
|
|
)
|
|
|
|
|
|
def test_safe_to_json_falls_back_on_lone_surrogates():
|
|
"""`safe_to_json` returns `to_json` output normally and escapes lone surrogates on the fallback."""
|
|
from pydantic_ai._instrumentation import safe_to_json
|
|
|
|
assert safe_to_json({'a': [1, 'b']}) == snapshot(b'{"a":[1,"b"]}')
|
|
assert safe_to_json('x\udce4y') == snapshot(b'"x\\udce4y"')
|
|
|
|
|
|
def test_instrumentation_settings_rejects_removed_version():
|
|
with pytest.raises(ValueError, match='Instrumentation version must be one of 2, 3, 4, or 5'):
|
|
InstrumentationSettings(version=1) # pyright: ignore[reportArgumentType]
|
|
|
|
|
|
@pytest.mark.parametrize('version', [2, 3, 4])
|
|
def test_instrumentation_settings_warns_for_deprecated_versions(version: Literal[2, 3, 4]):
|
|
settings = deprecated_instrumentation_settings(version=version)
|
|
assert settings.version == version
|
|
|
|
|
|
def test_instrumentation_settings_current_version_does_not_warn(recwarn: pytest.WarningsRecorder):
|
|
InstrumentationSettings()
|
|
InstrumentationSettings(version=5)
|
|
deprecation_warnings = [w for w in recwarn if issubclass(w.category, PydanticAIDeprecationWarning)]
|
|
assert deprecation_warnings == []
|
|
|
|
|
|
async def test_instrumented_model_stream(capfire: CaptureLogfire):
|
|
model = InstrumentedModel(MyModel(), InstrumentationSettings())
|
|
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart('user_prompt'),
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
async with model.request_stream(
|
|
messages,
|
|
model_settings=ModelSettings(temperature=1),
|
|
model_request_parameters=ModelRequestParameters(
|
|
function_tools=[],
|
|
allow_text_output=True,
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
),
|
|
) as response_stream:
|
|
assert [event async for event in response_stream] == snapshot(
|
|
[
|
|
PartStartEvent(index=0, part=TextPart(content='text1')),
|
|
FinalResultEvent(tool_name=None, tool_call_id=None),
|
|
PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='text2')),
|
|
PartEndEvent(index=0, part=TextPart(content='text1text2')),
|
|
]
|
|
)
|
|
|
|
assert capfire.exporter.exported_spans_as_dict(parse_json_attributes=True) == snapshot(
|
|
[
|
|
{
|
|
'name': 'chat gpt-4o',
|
|
'context': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
|
|
'parent': None,
|
|
'start_time': 1000000000,
|
|
'end_time': 2000000000,
|
|
'attributes': {
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'server.address': 'example.com',
|
|
'server.port': 8000,
|
|
'model_request_parameters': {
|
|
'function_tools': [],
|
|
'native_tools': [],
|
|
'output_mode': 'text',
|
|
'output_object': None,
|
|
'output_tools': [],
|
|
'prompted_output_template': None,
|
|
'allow_text_output': True,
|
|
'allow_image_output': False,
|
|
'instruction_parts': None,
|
|
'thinking': None,
|
|
},
|
|
'logfire.json_schema': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'gen_ai.input.messages': {'type': 'array'},
|
|
'gen_ai.output.messages': {'type': 'array'},
|
|
'model_request_parameters': {'type': 'object'},
|
|
},
|
|
},
|
|
'gen_ai.request.temperature': 1,
|
|
'logfire.msg': 'chat gpt-4o',
|
|
'gen_ai.input.messages': [{'role': 'user', 'parts': [{'type': 'text', 'content': 'user_prompt'}]}],
|
|
'gen_ai.output.messages': [
|
|
{'role': 'assistant', 'parts': [{'type': 'text', 'content': 'text1text2'}]}
|
|
],
|
|
'logfire.span_type': 'span',
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
'gen_ai.usage.input_tokens': 300,
|
|
'gen_ai.usage.output_tokens': 400,
|
|
'operation.cost': 0.00475,
|
|
'gen_ai.client.operation.time_to_first_chunk': IsFloat(),
|
|
},
|
|
},
|
|
]
|
|
)
|
|
|
|
assert capfire.log_exporter.exported_logs_as_dicts() == snapshot([])
|
|
|
|
# Streaming records the time-to-first-chunk histogram (value is non-deterministic, so
|
|
# assert shape rather than snapshot the float).
|
|
ttft_metrics = [
|
|
m for m in capfire.get_collected_metrics() if m['name'] == 'gen_ai.client.operation.time_to_first_chunk'
|
|
]
|
|
assert len(ttft_metrics) == 1
|
|
assert ttft_metrics[0]['unit'] == 's'
|
|
assert len(ttft_metrics[0]['data']['data_points']) == 1
|
|
|
|
|
|
async def test_instrumented_model_stream_break(capfire: CaptureLogfire):
|
|
model = InstrumentedModel(MyModel(), InstrumentationSettings())
|
|
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart('user_prompt'),
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
|
|
with pytest.raises(RuntimeError):
|
|
async with model.request_stream(
|
|
messages,
|
|
model_settings=ModelSettings(temperature=1),
|
|
model_request_parameters=ModelRequestParameters(
|
|
function_tools=[],
|
|
allow_text_output=True,
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
),
|
|
) as response_stream:
|
|
async for event in response_stream: # pragma: no branch
|
|
assert event == PartStartEvent(index=0, part=TextPart(content='text1'))
|
|
raise RuntimeError
|
|
|
|
assert capfire.exporter.exported_spans_as_dict(parse_json_attributes=True) == snapshot(
|
|
[
|
|
{
|
|
'name': 'chat gpt-4o',
|
|
'context': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
|
|
'parent': None,
|
|
'start_time': 1000000000,
|
|
'end_time': 3000000000,
|
|
'attributes': {
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'server.address': 'example.com',
|
|
'server.port': 8000,
|
|
'model_request_parameters': {
|
|
'function_tools': [],
|
|
'native_tools': [],
|
|
'output_mode': 'text',
|
|
'output_object': None,
|
|
'output_tools': [],
|
|
'prompted_output_template': None,
|
|
'allow_text_output': True,
|
|
'allow_image_output': False,
|
|
'instruction_parts': None,
|
|
'thinking': None,
|
|
},
|
|
'logfire.json_schema': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'gen_ai.input.messages': {'type': 'array'},
|
|
'gen_ai.output.messages': {'type': 'array'},
|
|
'model_request_parameters': {'type': 'object'},
|
|
},
|
|
},
|
|
'gen_ai.request.temperature': 1,
|
|
'logfire.msg': 'chat gpt-4o',
|
|
'gen_ai.input.messages': [{'role': 'user', 'parts': [{'type': 'text', 'content': 'user_prompt'}]}],
|
|
'gen_ai.output.messages': [{'role': 'assistant', 'parts': [{'type': 'text', 'content': 'text1'}]}],
|
|
'logfire.span_type': 'span',
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
'gen_ai.usage.input_tokens': 300,
|
|
'gen_ai.usage.output_tokens': 400,
|
|
'operation.cost': 0.00475,
|
|
'gen_ai.client.operation.time_to_first_chunk': IsFloat(),
|
|
'logfire.exception.fingerprint': '0000000000000000000000000000000000000000000000000000000000000000',
|
|
'logfire.level_num': 17,
|
|
},
|
|
'events': [
|
|
{
|
|
'name': 'exception',
|
|
'timestamp': 2000000000,
|
|
'attributes': {
|
|
'exception.type': 'RuntimeError',
|
|
'exception.message': '',
|
|
'exception.stacktrace': 'RuntimeError',
|
|
'exception.escaped': 'False',
|
|
},
|
|
}
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
assert capfire.log_exporter.exported_logs_as_dicts() == snapshot([])
|
|
|
|
|
|
async def test_instrumented_model_attributes_mode(capfire: CaptureLogfire):
|
|
model = InstrumentedModel(MyModel(), InstrumentationSettings())
|
|
assert model.system == 'openai'
|
|
assert model.model_name == 'gpt-4o'
|
|
|
|
messages = [
|
|
ModelRequest(
|
|
instructions='instructions',
|
|
parts=[
|
|
SystemPromptPart('system_prompt'),
|
|
UserPromptPart('user_prompt'),
|
|
ToolReturnPart('tool3', 'tool_return_content', 'tool_call_3'),
|
|
RetryPromptPart('retry_prompt1', tool_name='tool4', tool_call_id='tool_call_4'),
|
|
RetryPromptPart('retry_prompt2'),
|
|
{}, # test unexpected parts # type: ignore
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelResponse(parts=[TextPart('text3')]),
|
|
]
|
|
await model.request(
|
|
messages,
|
|
model_settings=ModelSettings(temperature=1),
|
|
model_request_parameters=ModelRequestParameters(
|
|
function_tools=[],
|
|
allow_text_output=True,
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
),
|
|
)
|
|
|
|
assert capfire.exporter.exported_spans_as_dict(parse_json_attributes=True) == snapshot(
|
|
[
|
|
{
|
|
'name': 'chat gpt-4o',
|
|
'context': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
|
|
'parent': None,
|
|
'start_time': 1000000000,
|
|
'end_time': 2000000000,
|
|
'attributes': {
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'server.address': 'example.com',
|
|
'server.port': 8000,
|
|
'model_request_parameters': {
|
|
'function_tools': [],
|
|
'native_tools': [],
|
|
'output_mode': 'text',
|
|
'output_object': None,
|
|
'output_tools': [],
|
|
'prompted_output_template': None,
|
|
'allow_text_output': True,
|
|
'allow_image_output': False,
|
|
'instruction_parts': None,
|
|
'thinking': None,
|
|
},
|
|
'gen_ai.request.temperature': 1,
|
|
'logfire.msg': 'chat gpt-4o',
|
|
'logfire.span_type': 'span',
|
|
'gen_ai.input.messages': [
|
|
{
|
|
'role': 'system',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'system_prompt'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt'},
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'tool_call_3',
|
|
'name': 'tool3',
|
|
'result': 'tool_return_content',
|
|
},
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'tool_call_4',
|
|
'name': 'tool4',
|
|
'result': """\
|
|
retry_prompt1
|
|
|
|
Fix the errors and try again.\
|
|
""",
|
|
},
|
|
{
|
|
'type': 'text',
|
|
'content': """\
|
|
Validation feedback:
|
|
retry_prompt2
|
|
|
|
Fix the errors and try again.\
|
|
""",
|
|
},
|
|
],
|
|
},
|
|
{'role': 'assistant', 'parts': [{'type': 'text', 'content': 'text3'}]},
|
|
],
|
|
'gen_ai.output.messages': [
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text1'},
|
|
{'type': 'tool_call', 'id': 'tool_call_1', 'name': 'tool1', 'arguments': 'args1'},
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'tool_call_2',
|
|
'name': 'tool2',
|
|
'arguments': {'args2': 3},
|
|
},
|
|
{'type': 'text', 'content': 'text2'},
|
|
],
|
|
}
|
|
],
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
'gen_ai.system_instructions': [{'type': 'text', 'content': 'instructions'}],
|
|
'gen_ai.usage.input_tokens': 100,
|
|
'gen_ai.usage.output_tokens': 200,
|
|
'gen_ai.usage.cache_creation.input_tokens': 10,
|
|
'gen_ai.usage.cache_read.input_tokens': 20,
|
|
'gen_ai.usage.details.reasoning_tokens': 30,
|
|
'gen_ai.usage.details.cache_write_tokens': 10,
|
|
'gen_ai.usage.details.cache_read_tokens': 20,
|
|
'gen_ai.usage.details.input_audio_tokens': 10,
|
|
'gen_ai.usage.details.cache_audio_read_tokens': 5,
|
|
'gen_ai.usage.details.output_audio_tokens': 30,
|
|
'logfire.json_schema': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'gen_ai.input.messages': {'type': 'array'},
|
|
'gen_ai.output.messages': {'type': 'array'},
|
|
'gen_ai.system_instructions': {'type': 'array'},
|
|
'model_request_parameters': {'type': 'object'},
|
|
},
|
|
},
|
|
'operation.cost': 0.002225,
|
|
'gen_ai.response.id': 'response_id',
|
|
},
|
|
},
|
|
]
|
|
)
|
|
|
|
assert capfire.get_collected_metrics() == snapshot(
|
|
[
|
|
{
|
|
'name': 'gen_ai.client.token.usage',
|
|
'description': 'Measures number of input and output tokens used',
|
|
'unit': '{token}',
|
|
'data': {
|
|
'data_points': [
|
|
{
|
|
'attributes': {
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
'gen_ai.token.type': 'input',
|
|
},
|
|
'start_time_unix_nano': IsInt(),
|
|
'time_unix_nano': IsInt(),
|
|
'count': 1,
|
|
'sum': 100,
|
|
'scale': 20,
|
|
'zero_count': 0,
|
|
'positive': {'offset': 6966588, 'bucket_counts': [1]},
|
|
'negative': {'offset': 0, 'bucket_counts': [0]},
|
|
'flags': 0,
|
|
'min': 100,
|
|
'max': 100,
|
|
'exemplars': [],
|
|
},
|
|
{
|
|
'attributes': {
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
'gen_ai.token.type': 'output',
|
|
},
|
|
'start_time_unix_nano': IsInt(),
|
|
'time_unix_nano': IsInt(),
|
|
'count': 1,
|
|
'sum': 200,
|
|
'scale': 20,
|
|
'zero_count': 0,
|
|
'positive': {'offset': 8015164, 'bucket_counts': [1]},
|
|
'negative': {'offset': 0, 'bucket_counts': [0]},
|
|
'flags': 0,
|
|
'min': 200,
|
|
'max': 200,
|
|
'exemplars': [],
|
|
},
|
|
],
|
|
'aggregation_temporality': 1,
|
|
},
|
|
},
|
|
{
|
|
'name': 'operation.cost',
|
|
'description': 'Monetary cost',
|
|
'unit': '{USD}',
|
|
'data': {
|
|
'data_points': [
|
|
{
|
|
'attributes': {
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
},
|
|
'start_time_unix_nano': IsInt(),
|
|
'time_unix_nano': IsInt(),
|
|
'count': 1,
|
|
'sum': 0.002225,
|
|
'scale': 20,
|
|
'zero_count': 0,
|
|
'positive': {'offset': -9240030, 'bucket_counts': [1]},
|
|
'negative': {'offset': 0, 'bucket_counts': [0]},
|
|
'flags': 0,
|
|
'min': 0.002225,
|
|
'max': 0.002225,
|
|
'exemplars': [],
|
|
}
|
|
],
|
|
'aggregation_temporality': 1,
|
|
},
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_message_parts_compaction_part():
|
|
"""CompactionPart is skipped in otel_message_parts (not a standard GenAI convention type)."""
|
|
from pydantic_ai.messages import CompactionPart
|
|
|
|
messages: list[ModelMessage] = [
|
|
ModelResponse(parts=[CompactionPart(content='Summary.', provider_name='anthropic'), TextPart('response')]),
|
|
]
|
|
settings = InstrumentationSettings()
|
|
otel_messages = settings.messages_to_otel_messages(messages)
|
|
# CompactionPart is skipped; only TextPart appears
|
|
assert otel_messages == snapshot([{'role': 'assistant', 'parts': [{'type': 'text', 'content': 'response'}]}])
|
|
|
|
|
|
def test_messages_to_otel_messages_multimodal_v3(document_content: BinaryContent):
|
|
"""Test that version 3 keeps the pre-v4 multimodal format."""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['user_prompt', ImageUrl('https://example.com/image.png')])],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['user_prompt2', AudioUrl('https://example.com/audio.mp3')])],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['user_prompt3', DocumentUrl('https://example.com/document.pdf')])],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['user_prompt4', VideoUrl('https://example.com/video.mp4')])],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'user_prompt5',
|
|
ImageUrl('https://example.com/image2.png'),
|
|
AudioUrl('https://example.com/audio2.mp3'),
|
|
DocumentUrl('https://example.com/document2.pdf'),
|
|
VideoUrl('https://example.com/video2.mp4'),
|
|
]
|
|
)
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelRequest(parts=[UserPromptPart(content=['user_prompt6', document_content])], timestamp=IsDatetime()),
|
|
ModelResponse(parts=[TextPart('text1')]),
|
|
ModelResponse(parts=[FilePart(content=document_content)]),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=3)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt'},
|
|
{'type': 'image-url', 'url': 'https://example.com/image.png'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt2'},
|
|
{'type': 'audio-url', 'url': 'https://example.com/audio.mp3'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt3'},
|
|
{'type': 'document-url', 'url': 'https://example.com/document.pdf'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt4'},
|
|
{'type': 'video-url', 'url': 'https://example.com/video.mp4'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt5'},
|
|
{'type': 'image-url', 'url': 'https://example.com/image2.png'},
|
|
{'type': 'audio-url', 'url': 'https://example.com/audio2.mp3'},
|
|
{'type': 'document-url', 'url': 'https://example.com/document2.pdf'},
|
|
{'type': 'video-url', 'url': 'https://example.com/video2.mp4'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt6'},
|
|
{
|
|
'type': 'binary',
|
|
'media_type': 'application/pdf',
|
|
'content': IsStr(),
|
|
},
|
|
],
|
|
},
|
|
{'role': 'assistant', 'parts': [{'type': 'text', 'content': 'text1'}]},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'binary',
|
|
'media_type': 'application/pdf',
|
|
'content': IsStr(),
|
|
}
|
|
],
|
|
},
|
|
]
|
|
)
|
|
settings_without_binary = deprecated_instrumentation_settings(version=3, include_binary_content=False)
|
|
assert settings_without_binary.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt'},
|
|
{'type': 'image-url', 'url': 'https://example.com/image.png'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt2'},
|
|
{'type': 'audio-url', 'url': 'https://example.com/audio.mp3'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt3'},
|
|
{'type': 'document-url', 'url': 'https://example.com/document.pdf'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt4'},
|
|
{'type': 'video-url', 'url': 'https://example.com/video.mp4'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt5'},
|
|
{'type': 'image-url', 'url': 'https://example.com/image2.png'},
|
|
{'type': 'audio-url', 'url': 'https://example.com/audio2.mp3'},
|
|
{'type': 'document-url', 'url': 'https://example.com/document2.pdf'},
|
|
{'type': 'video-url', 'url': 'https://example.com/video2.mp4'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'user_prompt6'},
|
|
{'type': 'binary', 'media_type': 'application/pdf'},
|
|
],
|
|
},
|
|
{'role': 'assistant', 'parts': [{'type': 'text', 'content': 'text1'}]},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'binary', 'media_type': 'application/pdf'},
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_multimodal_v4():
|
|
"""Test that version 4 uses GenAI semantic conventions for multimodal inputs."""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'Describe these files',
|
|
ImageUrl('https://example.com/image.jpg', media_type='image/jpeg'),
|
|
AudioUrl('https://example.com/audio.mp3', media_type='audio/mpeg'),
|
|
DocumentUrl('https://example.com/doc.pdf', media_type='application/pdf'),
|
|
VideoUrl('https://example.com/video.mp4', media_type='video/mp4'),
|
|
]
|
|
)
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Describe these files'},
|
|
{
|
|
'type': 'uri',
|
|
'modality': 'image',
|
|
'uri': 'https://example.com/image.jpg',
|
|
'mime_type': 'image/jpeg',
|
|
},
|
|
{
|
|
'type': 'uri',
|
|
'modality': 'audio',
|
|
'uri': 'https://example.com/audio.mp3',
|
|
'mime_type': 'audio/mpeg',
|
|
},
|
|
{
|
|
'type': 'uri',
|
|
'uri': 'https://example.com/doc.pdf',
|
|
'mime_type': 'application/pdf',
|
|
},
|
|
{
|
|
'type': 'uri',
|
|
'modality': 'video',
|
|
'uri': 'https://example.com/video.mp4',
|
|
'mime_type': 'video/mp4',
|
|
},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_multimodal_v4_no_content():
|
|
"""Test that version 4 with include_content=False omits uri but keeps mime_type."""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'Describe this',
|
|
ImageUrl('https://example.com/image.jpg', media_type='image/jpeg'),
|
|
]
|
|
)
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4, include_content=False)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text'},
|
|
{'type': 'uri', 'modality': 'image', 'mime_type': 'image/jpeg'},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_binary_content_v4():
|
|
"""Test that version 4 uses blob format with modality for BinaryContent."""
|
|
image_data = BinaryContent(data=b'fake image data', media_type='image/png')
|
|
audio_data = BinaryContent(data=b'fake audio data', media_type='audio/mpeg')
|
|
video_data = BinaryContent(data=b'fake video data', media_type='video/mp4')
|
|
doc_data = BinaryContent(data=b'fake doc data', media_type='application/pdf')
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'Analyze these files',
|
|
image_data,
|
|
audio_data,
|
|
video_data,
|
|
doc_data,
|
|
]
|
|
)
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Analyze these files'},
|
|
{
|
|
'type': 'blob',
|
|
'modality': 'image',
|
|
'mime_type': 'image/png',
|
|
'content': image_data.base64,
|
|
},
|
|
{
|
|
'type': 'blob',
|
|
'modality': 'audio',
|
|
'mime_type': 'audio/mpeg',
|
|
'content': audio_data.base64,
|
|
},
|
|
{
|
|
'type': 'blob',
|
|
'modality': 'video',
|
|
'mime_type': 'video/mp4',
|
|
'content': video_data.base64,
|
|
},
|
|
{
|
|
'type': 'blob',
|
|
'mime_type': 'application/pdf',
|
|
'content': doc_data.base64,
|
|
},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_binary_content_v4_no_content():
|
|
"""Test that version 4 with include_content=False omits content but keeps mime_type."""
|
|
image_data = BinaryContent(data=b'fake image data', media_type='image/png')
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['Analyze this', image_data])],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4, include_content=False)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text'},
|
|
{'type': 'blob', 'modality': 'image', 'mime_type': 'image/png'},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_url_without_extension_v4():
|
|
"""Test that version 4 gracefully handles URLs where media_type cannot be inferred."""
|
|
# URL without extension - media_type will raise ValueError
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'Describe this',
|
|
ImageUrl('https://example.com/image_no_extension'),
|
|
]
|
|
)
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Describe this'},
|
|
{
|
|
'type': 'uri',
|
|
'modality': 'image',
|
|
'uri': 'https://example.com/image_no_extension',
|
|
# Note: mime_type is omitted because it cannot be inferred
|
|
},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_without_content(document_content: BinaryContent):
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(parts=[SystemPromptPart('system_prompt')], timestamp=IsDatetime()),
|
|
ModelResponse(parts=[TextPart('text1')]),
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'user_prompt1',
|
|
VideoUrl('https://example.com/video.mp4'),
|
|
ImageUrl('https://example.com/image.png'),
|
|
AudioUrl('https://example.com/audio.mp3'),
|
|
DocumentUrl('https://example.com/document.pdf'),
|
|
document_content,
|
|
]
|
|
)
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelResponse(parts=[TextPart('text2'), ToolCallPart(tool_name='my_tool', args={'a': 13, 'b': 4})]),
|
|
ModelRequest(parts=[ToolReturnPart('tool', 'tool_return_content', 'tool_call_1')], timestamp=IsDatetime()),
|
|
ModelRequest(
|
|
parts=[RetryPromptPart('retry_prompt', tool_name='tool', tool_call_id='tool_call_2')],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
ModelRequest(parts=[UserPromptPart(content=['user_prompt2', document_content])], timestamp=IsDatetime()),
|
|
ModelRequest(parts=[UserPromptPart('simple text prompt')], timestamp=IsDatetime()),
|
|
ModelResponse(parts=[FilePart(content=document_content)]),
|
|
]
|
|
settings = InstrumentationSettings(include_content=False)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{'role': 'system', 'parts': [{'type': 'text'}]},
|
|
{'role': 'assistant', 'parts': [{'type': 'text'}]},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text'},
|
|
{'type': 'uri', 'modality': 'video', 'mime_type': 'video/mp4'},
|
|
{'type': 'uri', 'modality': 'image', 'mime_type': 'image/png'},
|
|
{'type': 'uri', 'modality': 'audio', 'mime_type': 'audio/mpeg'},
|
|
{'type': 'uri', 'mime_type': 'application/pdf'},
|
|
{'type': 'blob', 'mime_type': 'application/pdf'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'text'},
|
|
{'type': 'tool_call', 'id': IsStr(), 'name': 'my_tool'},
|
|
],
|
|
},
|
|
{'role': 'user', 'parts': [{'type': 'tool_call_response', 'id': 'tool_call_1', 'name': 'tool'}]},
|
|
{'role': 'user', 'parts': [{'type': 'tool_call_response', 'id': 'tool_call_2', 'name': 'tool'}]},
|
|
{'role': 'user', 'parts': [{'type': 'text'}, {'type': 'blob', 'mime_type': 'application/pdf'}]},
|
|
{'role': 'user', 'parts': [{'type': 'text'}]},
|
|
{'role': 'assistant', 'parts': [{'type': 'blob', 'mime_type': 'application/pdf'}]},
|
|
]
|
|
)
|
|
|
|
|
|
def test_message_with_thinking_parts():
|
|
messages: list[ModelMessage] = [
|
|
ModelResponse(parts=[TextPart('text1'), ThinkingPart('thinking1'), TextPart('text2')]),
|
|
ModelResponse(parts=[ThinkingPart('thinking2')]),
|
|
ModelResponse(parts=[ThinkingPart('thinking3'), TextPart('text3')]),
|
|
]
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text1'},
|
|
{'type': 'thinking', 'content': 'thinking1'},
|
|
{'type': 'text', 'content': 'text2'},
|
|
],
|
|
},
|
|
{'role': 'assistant', 'parts': [{'type': 'thinking', 'content': 'thinking2'}]},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [{'type': 'thinking', 'content': 'thinking3'}, {'type': 'text', 'content': 'text3'}],
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
async def test_response_cost_error(capfire: CaptureLogfire, monkeypatch: pytest.MonkeyPatch):
|
|
model = InstrumentedModel(MyModel())
|
|
|
|
messages: list[ModelMessage] = [ModelRequest(parts=[UserPromptPart('user_prompt')], timestamp=IsDatetime())]
|
|
monkeypatch.setattr(ModelResponse, 'cost', None)
|
|
|
|
with warns(
|
|
snapshot(
|
|
[
|
|
"CostCalculationFailedWarning: Failed to get cost from response: TypeError: 'NoneType' object is not callable"
|
|
]
|
|
)
|
|
):
|
|
await model.request(messages, model_settings=ModelSettings(), model_request_parameters=ModelRequestParameters())
|
|
|
|
assert capfire.exporter.exported_spans_as_dict(parse_json_attributes=True) == snapshot(
|
|
[
|
|
{
|
|
'name': 'chat gpt-4o',
|
|
'context': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
|
|
'parent': None,
|
|
'start_time': 1000000000,
|
|
'end_time': 2000000000,
|
|
'attributes': {
|
|
'gen_ai.operation.name': 'chat',
|
|
'gen_ai.provider.name': 'openai',
|
|
'gen_ai.system': 'openai',
|
|
'gen_ai.request.model': 'gpt-4o',
|
|
'server.address': 'example.com',
|
|
'server.port': 8000,
|
|
'model_request_parameters': {
|
|
'function_tools': [],
|
|
'native_tools': [],
|
|
'output_mode': 'text',
|
|
'output_object': None,
|
|
'output_tools': [],
|
|
'prompted_output_template': None,
|
|
'allow_text_output': True,
|
|
'allow_image_output': False,
|
|
'instruction_parts': None,
|
|
'thinking': None,
|
|
},
|
|
'logfire.span_type': 'span',
|
|
'logfire.msg': 'chat gpt-4o',
|
|
'gen_ai.input.messages': [{'role': 'user', 'parts': [{'type': 'text', 'content': 'user_prompt'}]}],
|
|
'gen_ai.output.messages': [
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text1'},
|
|
{'type': 'tool_call', 'id': 'tool_call_1', 'name': 'tool1', 'arguments': 'args1'},
|
|
{'type': 'tool_call', 'id': 'tool_call_2', 'name': 'tool2', 'arguments': {'args2': 3}},
|
|
{'type': 'text', 'content': 'text2'},
|
|
],
|
|
}
|
|
],
|
|
'logfire.json_schema': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'gen_ai.input.messages': {'type': 'array'},
|
|
'gen_ai.output.messages': {'type': 'array'},
|
|
'model_request_parameters': {'type': 'object'},
|
|
},
|
|
},
|
|
'gen_ai.usage.input_tokens': 100,
|
|
'gen_ai.usage.output_tokens': 200,
|
|
'gen_ai.usage.cache_creation.input_tokens': 10,
|
|
'gen_ai.usage.cache_read.input_tokens': 20,
|
|
'gen_ai.usage.details.reasoning_tokens': 30,
|
|
'gen_ai.usage.details.cache_write_tokens': 10,
|
|
'gen_ai.usage.details.cache_read_tokens': 20,
|
|
'gen_ai.usage.details.input_audio_tokens': 10,
|
|
'gen_ai.usage.details.cache_audio_read_tokens': 5,
|
|
'gen_ai.usage.details.output_audio_tokens': 30,
|
|
'gen_ai.response.model': 'gpt-4o-2024-11-20',
|
|
'gen_ai.response.id': 'response_id',
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_message_with_native_tool_calls():
|
|
messages: list[ModelMessage] = [
|
|
ModelResponse(
|
|
parts=[
|
|
TextPart('text1'),
|
|
NativeToolCallPart('code_execution', {'code': '2 * 2'}, tool_call_id='tool_call_1'),
|
|
NativeToolReturnPart('code_execution', {'output': '4'}, tool_call_id='tool_call_1'),
|
|
TextPart('text2'),
|
|
NativeToolCallPart(
|
|
'web_search',
|
|
'{"query": "weather: San Francisco, CA", "type": "search"}',
|
|
tool_call_id='tool_call_2',
|
|
),
|
|
NativeToolReturnPart(
|
|
'web_search',
|
|
[
|
|
{
|
|
'url': 'https://www.weather.com/weather/today/l/USCA0987:1:US',
|
|
'title': 'Weather in San Francisco',
|
|
}
|
|
],
|
|
tool_call_id='tool_call_2',
|
|
),
|
|
TextPart('text3'),
|
|
]
|
|
),
|
|
]
|
|
settings = InstrumentationSettings()
|
|
# Built-in tool calls are only included in v2-style messages, not v1-style events,
|
|
# as the spec does not yet allow tool results coming from the assistant,
|
|
# and Logfire has special handling for the `type='tool_call_response', 'builtin=True'` messages, but not events.
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text1'},
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'tool_call_1',
|
|
'name': 'code_execution',
|
|
'builtin': True,
|
|
'arguments': {'code': '2 * 2'},
|
|
},
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'tool_call_1',
|
|
'name': 'code_execution',
|
|
'builtin': True,
|
|
'result': {'output': '4'},
|
|
},
|
|
{'type': 'text', 'content': 'text2'},
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'tool_call_2',
|
|
'name': 'web_search',
|
|
'builtin': True,
|
|
'arguments': '{"query": "weather: San Francisco, CA", "type": "search"}',
|
|
},
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'tool_call_2',
|
|
'name': 'web_search',
|
|
'builtin': True,
|
|
'result': [
|
|
{
|
|
'url': 'https://www.weather.com/weather/today/l/USCA0987:1:US',
|
|
'title': 'Weather in San Francisco',
|
|
}
|
|
],
|
|
},
|
|
{'type': 'text', 'content': 'text3'},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_cache_point_in_user_prompt():
|
|
"""Test that CachePoint is correctly skipped in OpenTelemetry conversion.
|
|
|
|
CachePoint is a marker for prompt caching and should not be included in the
|
|
OpenTelemetry message parts output.
|
|
"""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['text before', CachePoint(), 'text after'])], timestamp=IsDatetime()
|
|
),
|
|
]
|
|
settings = InstrumentationSettings()
|
|
|
|
# Test otel_message_parts - CachePoint should be skipped
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text before'},
|
|
{'type': 'text', 'content': 'text after'},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
# Test with multiple CachePoints
|
|
messages_multi: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(content=['first', CachePoint(), 'second', CachePoint(), 'third']),
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
assert settings.messages_to_otel_messages(messages_multi) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'first'},
|
|
{'type': 'text', 'content': 'second'},
|
|
{'type': 'text', 'content': 'third'},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
# Test with CachePoint mixed with other content types
|
|
messages_mixed: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'context',
|
|
CachePoint(),
|
|
ImageUrl('https://example.com/image.jpg'),
|
|
CachePoint(),
|
|
'question',
|
|
]
|
|
),
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
assert settings.messages_to_otel_messages(messages_mixed) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'context'},
|
|
{
|
|
'type': 'uri',
|
|
'modality': 'image',
|
|
'mime_type': 'image/jpeg',
|
|
'uri': 'https://example.com/image.jpg',
|
|
},
|
|
{'type': 'text', 'content': 'question'},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_build_tool_definitions():
|
|
"""Test build_tool_definitions with various tool configurations."""
|
|
from pydantic_ai._instrumentation import build_tool_definitions
|
|
from pydantic_ai.tools import ToolDefinition
|
|
|
|
tool_without_params = ToolDefinition(
|
|
name='no_params_tool',
|
|
description='A tool without parameters',
|
|
parameters_json_schema={},
|
|
)
|
|
|
|
tool_with_params = ToolDefinition(
|
|
name='with_params_tool',
|
|
description='A tool with parameters',
|
|
parameters_json_schema={'type': 'object', 'properties': {'x': {'type': 'integer'}}},
|
|
)
|
|
|
|
tool_no_description = ToolDefinition(
|
|
name='no_desc_tool',
|
|
description=None,
|
|
parameters_json_schema={'type': 'object', 'properties': {}},
|
|
)
|
|
|
|
params = ModelRequestParameters(
|
|
function_tools=[tool_without_params, tool_with_params, tool_no_description],
|
|
native_tools=[],
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
prompted_output_template=None,
|
|
allow_text_output=True,
|
|
allow_image_output=False,
|
|
)
|
|
|
|
result = build_tool_definitions(params)
|
|
|
|
assert result == [
|
|
{'type': 'function', 'name': 'no_params_tool', 'description': 'A tool without parameters'},
|
|
{
|
|
'type': 'function',
|
|
'name': 'with_params_tool',
|
|
'description': 'A tool with parameters',
|
|
'parameters': {'type': 'object', 'properties': {'x': {'type': 'integer'}}},
|
|
},
|
|
{
|
|
'type': 'function',
|
|
'name': 'no_desc_tool',
|
|
'parameters': {'type': 'object', 'properties': {}},
|
|
},
|
|
]
|
|
|
|
|
|
def test_annotate_tool_call_otel_metadata():
|
|
"""`annotate_tool_call_otel_metadata` copies metadata from tool defs onto matching tool call parts."""
|
|
from pydantic_ai._instrumentation import annotate_tool_call_otel_metadata
|
|
from pydantic_ai.tools import ToolDefinition
|
|
|
|
response = ModelResponse(
|
|
parts=[
|
|
ToolCallPart(tool_name='run_code_with_tools', args={'code': 'print("hi")'}, tool_call_id='call_1'),
|
|
ToolCallPart(tool_name='other_tool', args={'x': 1}, tool_call_id='call_2'),
|
|
ToolCallPart(tool_name='unrelated_metadata_tool', args={'y': 1}, tool_call_id='call_3'),
|
|
TextPart('some text'),
|
|
]
|
|
)
|
|
|
|
params = ModelRequestParameters(
|
|
function_tools=[
|
|
ToolDefinition(
|
|
name='run_code_with_tools',
|
|
parameters_json_schema={'type': 'object', 'properties': {}},
|
|
metadata={'code_arg_name': 'code', 'code_arg_language': 'python'},
|
|
),
|
|
ToolDefinition(
|
|
name='other_tool',
|
|
parameters_json_schema={'type': 'object', 'properties': {}},
|
|
),
|
|
# Truthy metadata without `code_arg_*` keys exercises the branches that skip each
|
|
# individual `if code_arg_name`/`if code_arg_language`/`if otel_metadata` check.
|
|
ToolDefinition(
|
|
name='unrelated_metadata_tool',
|
|
parameters_json_schema={'type': 'object', 'properties': {}},
|
|
metadata={'foo': 'bar'},
|
|
),
|
|
],
|
|
native_tools=[],
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
prompted_output_template=None,
|
|
allow_text_output=True,
|
|
allow_image_output=False,
|
|
)
|
|
|
|
annotate_tool_call_otel_metadata(response, params)
|
|
|
|
code_part = response.parts[0]
|
|
assert isinstance(code_part, ToolCallPart)
|
|
assert code_part.otel_metadata == {'code_arg_name': 'code', 'code_arg_language': 'python'}
|
|
|
|
other_part = response.parts[1]
|
|
assert isinstance(other_part, ToolCallPart)
|
|
assert other_part.otel_metadata is None
|
|
|
|
unrelated_part = response.parts[2]
|
|
assert isinstance(unrelated_part, ToolCallPart)
|
|
assert unrelated_part.otel_metadata is None
|
|
|
|
|
|
def test_builtin_code_execution_otel_metadata_in_otel_messages():
|
|
"""Builtin code execution tool calls carry code_arg metadata in OTel output."""
|
|
call_part = NativeToolCallPart(
|
|
tool_name='code_execution', args={'code': '2 * 2'}, tool_call_id='call_1', provider_name='anthropic'
|
|
)
|
|
call_part.otel_metadata = {'code_arg_name': 'code', 'code_arg_language': 'python'}
|
|
|
|
messages: list[ModelMessage] = [ModelResponse(parts=[call_part])]
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'call_1',
|
|
'name': 'code_execution',
|
|
'builtin': True,
|
|
'code_arg_name': 'code',
|
|
'code_arg_language': 'python',
|
|
'arguments': {'code': '2 * 2'},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_builtin_code_execution_snippet_arg():
|
|
"""Bedrock's 'snippet' arg name is preserved in OTel output."""
|
|
call_part = NativeToolCallPart(
|
|
tool_name='code_execution', args={'snippet': '1 + 1'}, tool_call_id='call_1', provider_name='bedrock'
|
|
)
|
|
call_part.otel_metadata = {'code_arg_name': 'snippet', 'code_arg_language': 'python'}
|
|
|
|
messages: list[ModelMessage] = [ModelResponse(parts=[call_part])]
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'call_1',
|
|
'name': 'code_execution',
|
|
'builtin': True,
|
|
'code_arg_name': 'snippet',
|
|
'code_arg_language': 'python',
|
|
'arguments': {'snippet': '1 + 1'},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_otel_metadata_in_otel_messages():
|
|
"""`otel_metadata` on a function tool call flows through to OTel message output."""
|
|
tool_call = ToolCallPart(tool_name='run_code_with_tools', args={'code': 'x = 1 + 2'}, tool_call_id='call_1')
|
|
tool_call.otel_metadata = {'code_arg_name': 'code', 'code_arg_language': 'python'}
|
|
|
|
messages: list[ModelMessage] = [ModelResponse(parts=[tool_call])]
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'call_1',
|
|
'name': 'run_code_with_tools',
|
|
'code_arg_name': 'code',
|
|
'code_arg_language': 'python',
|
|
'arguments': {'code': 'x = 1 + 2'},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_otel_metadata_partial_only_arg_name():
|
|
"""`otel_metadata` with only `code_arg_name` set surfaces just that key."""
|
|
tool_call = ToolCallPart(tool_name='run_code', args={'code': '1+1'}, tool_call_id='call_1')
|
|
tool_call.otel_metadata = {'code_arg_name': 'code'}
|
|
|
|
messages: list[ModelMessage] = [ModelResponse(parts=[tool_call])]
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'call_1',
|
|
'name': 'run_code',
|
|
'code_arg_name': 'code',
|
|
'arguments': {'code': '1+1'},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_otel_metadata_partial_only_arg_language():
|
|
"""`otel_metadata` with only `code_arg_language` set surfaces just that key."""
|
|
tool_call = ToolCallPart(tool_name='run_code', args={'code': '1+1'}, tool_call_id='call_1')
|
|
tool_call.otel_metadata = {'code_arg_language': 'python'}
|
|
|
|
messages: list[ModelMessage] = [ModelResponse(parts=[tool_call])]
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'call_1',
|
|
'name': 'run_code',
|
|
'code_arg_language': 'python',
|
|
'arguments': {'code': '1+1'},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_otel_metadata_not_present_without_annotation():
|
|
"""`code_arg_name`/`code_arg_language` are absent when `otel_metadata` is not set."""
|
|
messages: list[ModelMessage] = [
|
|
ModelResponse(parts=[ToolCallPart(tool_name='some_tool', args={'x': 1}, tool_call_id='call_1')]),
|
|
]
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'call_1',
|
|
'name': 'some_tool',
|
|
'arguments': {'x': 1},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_file_part_v4(document_content: BinaryContent):
|
|
"""Test that version 4 uses blob format for FilePart in ModelResponse (output messages)."""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(parts=[UserPromptPart(content='Generate a document')], timestamp=IsDatetime()),
|
|
ModelResponse(parts=[FilePart(content=document_content)]),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Generate a document'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{
|
|
'type': 'blob',
|
|
'mime_type': 'application/pdf',
|
|
'content': document_content.base64,
|
|
},
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_file_part_v4_no_content(document_content: BinaryContent):
|
|
"""Test that version 4 with include_content=False omits content but keeps mime_type for FilePart."""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(parts=[UserPromptPart(content='Generate a document')], timestamp=IsDatetime()),
|
|
ModelResponse(parts=[FilePart(content=document_content)]),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4, include_content=False)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'blob', 'mime_type': 'application/pdf'},
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_cache_point_v4():
|
|
"""Test that CachePoint is correctly skipped with version 4."""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[
|
|
UserPromptPart(
|
|
content=[
|
|
'text',
|
|
CachePoint(),
|
|
ImageUrl('https://example.com/image.jpg', media_type='image/jpeg'),
|
|
CachePoint(),
|
|
]
|
|
)
|
|
],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text'},
|
|
{
|
|
'type': 'uri',
|
|
'modality': 'image',
|
|
'mime_type': 'image/jpeg',
|
|
'uri': 'https://example.com/image.jpg',
|
|
},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_builtin_tool_v4():
|
|
"""Test that NativeToolCallPart works correctly with version 4."""
|
|
messages: list[ModelMessage] = [
|
|
ModelResponse(
|
|
parts=[
|
|
TextPart('text'),
|
|
NativeToolCallPart('code_execution', {'code': '2 * 2'}, tool_call_id='tool_call_1'),
|
|
NativeToolReturnPart('code_execution', {'output': '4'}, tool_call_id='tool_call_1'),
|
|
]
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'text'},
|
|
{
|
|
'type': 'tool_call',
|
|
'id': 'tool_call_1',
|
|
'name': 'code_execution',
|
|
'builtin': True,
|
|
'arguments': {'code': '2 * 2'},
|
|
},
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'tool_call_1',
|
|
'name': 'code_execution',
|
|
'builtin': True,
|
|
'result': {'output': '4'},
|
|
},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_binary_content_v4_no_binary():
|
|
"""Test version 4 with include_binary_content=False omits the content field entirely."""
|
|
image_data = BinaryContent(data=b'fake image data', media_type='image/png')
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['Analyze this', image_data])],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4, include_binary_content=False)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Analyze this'},
|
|
{'type': 'blob', 'modality': 'image', 'mime_type': 'image/png'},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_file_part_v4_no_binary(document_content: BinaryContent):
|
|
"""Test version 4 FilePart with include_binary_content=False omits the content field."""
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(parts=[UserPromptPart(content='Generate a document')], timestamp=IsDatetime()),
|
|
ModelResponse(parts=[FilePart(content=document_content)]),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4, include_binary_content=False)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Generate a document'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'blob', 'mime_type': 'application/pdf'},
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_binary_content_v4_unknown_modality():
|
|
"""Test version 4 with unknown media type (no modality field added)."""
|
|
unknown_data = BinaryContent(data=b'unknown data', media_type='x-custom/data')
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(
|
|
parts=[UserPromptPart(content=['Check this', unknown_data])],
|
|
timestamp=IsDatetime(),
|
|
),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Check this'},
|
|
{'type': 'blob', 'mime_type': 'x-custom/data', 'content': unknown_data.base64},
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_file_part_v4_unknown_modality():
|
|
"""Test version 4 FilePart with unknown media type (no modality field added)."""
|
|
unknown_content = BinaryContent(data=b'unknown file data', media_type='x-vendor/custom-format')
|
|
messages: list[ModelMessage] = [
|
|
ModelRequest(parts=[UserPromptPart(content='Process file')], timestamp=IsDatetime()),
|
|
ModelResponse(parts=[FilePart(content=unknown_content)]),
|
|
]
|
|
settings = deprecated_instrumentation_settings(version=4)
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{'type': 'text', 'content': 'Process file'},
|
|
],
|
|
},
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [
|
|
{'type': 'blob', 'mime_type': 'x-vendor/custom-format', 'content': unknown_content.base64},
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def test_messages_to_otel_messages_serialization_errors():
|
|
class Foo:
|
|
def __repr__(self) -> str:
|
|
return 'Foo()'
|
|
|
|
class Bar:
|
|
def __repr__(self) -> str:
|
|
raise ValueError('error!')
|
|
|
|
messages: list[ModelMessage] = [
|
|
ModelResponse(parts=[ToolCallPart('tool', {'arg': Foo()}, tool_call_id='tool_call_id')]),
|
|
ModelRequest(parts=[ToolReturnPart('tool', Bar(), tool_call_id='return_tool_call_id')], timestamp=IsDatetime()),
|
|
]
|
|
|
|
settings = InstrumentationSettings()
|
|
assert settings.messages_to_otel_messages(messages) == snapshot(
|
|
[
|
|
{
|
|
'role': 'assistant',
|
|
'parts': [{'type': 'tool_call', 'id': 'tool_call_id', 'name': 'tool', 'arguments': {'arg': 'Foo()'}}],
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'parts': [
|
|
{
|
|
'type': 'tool_call_response',
|
|
'id': 'return_tool_call_id',
|
|
'name': 'tool',
|
|
'result': 'Unable to serialize: error!',
|
|
}
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
async def test_instrumented_model_count_tokens(capfire: CaptureLogfire):
|
|
messages: list[ModelMessage] = [ModelRequest(parts=[UserPromptPart('Hello, world!')], timestamp=IsDatetime())]
|
|
model = InstrumentedModel(MyModel())
|
|
usage = await model.count_tokens(
|
|
messages, model_settings=ModelSettings(), model_request_parameters=ModelRequestParameters()
|
|
)
|
|
assert usage == RequestUsage(input_tokens=10)
|
|
|
|
|
|
async def test_instrumented_model_with_tools_and_finish_reason(capfire: CaptureLogfire):
|
|
"""Test _instrument() with tool definitions and a response that has finish_reason."""
|
|
from pydantic_ai.tools import ToolDefinition
|
|
|
|
class FinishReasonModel(MyModel):
|
|
async def request(
|
|
self,
|
|
messages: list[ModelMessage],
|
|
model_settings: ModelSettings | None,
|
|
model_request_parameters: ModelRequestParameters,
|
|
) -> ModelResponse:
|
|
return ModelResponse(
|
|
parts=[TextPart('done')],
|
|
usage=RequestUsage(input_tokens=10, output_tokens=5),
|
|
model_name='gpt-4o-2024-11-20',
|
|
provider_response_id='resp-123',
|
|
finish_reason='stop',
|
|
)
|
|
|
|
tool_def = ToolDefinition(
|
|
name='get_weather',
|
|
description='Get the weather',
|
|
parameters_json_schema={'type': 'object', 'properties': {'city': {'type': 'string'}}},
|
|
)
|
|
|
|
model = InstrumentedModel(FinishReasonModel())
|
|
messages: list[ModelMessage] = [ModelRequest(parts=[UserPromptPart('Hello')], timestamp=IsDatetime())]
|
|
await model.request(
|
|
messages,
|
|
model_settings=None,
|
|
model_request_parameters=ModelRequestParameters(
|
|
function_tools=[tool_def],
|
|
allow_text_output=True,
|
|
output_tools=[],
|
|
output_mode='text',
|
|
output_object=None,
|
|
),
|
|
)
|
|
|
|
spans = capfire.exporter.exported_spans_as_dict(parse_json_attributes=True)
|
|
assert len(spans) == 1
|
|
attrs = spans[0]['attributes']
|
|
# Tool definitions should be set
|
|
assert attrs['gen_ai.tool.definitions'] == snapshot(
|
|
[
|
|
{
|
|
'type': 'function',
|
|
'name': 'get_weather',
|
|
'description': 'Get the weather',
|
|
'parameters': {'type': 'object', 'properties': {'city': {'type': 'string'}}},
|
|
}
|
|
]
|
|
)
|
|
# finish_reason should be set
|
|
assert attrs['gen_ai.response.finish_reasons'] == ('stop',)
|
|
assert attrs['gen_ai.response.id'] == 'resp-123'
|
|
|
|
|
|
async def test_instrumented_model_tolerates_lone_surrogates_in_request_parameters(capfire: CaptureLogfire):
|
|
"""Lone surrogates in tool definitions / request parameters must not crash instrumentation.
|
|
|
|
`gen_ai.tool.definitions` and `model_request_parameters` are serialized on the model-request
|
|
span regardless of `include_content`; routing them through `safe_to_json` keeps a surrogate
|
|
(e.g. text decoded with `errors='surrogateescape'`) in a tool description from raising
|
|
`PydanticSerializationError` and crashing an otherwise-successful run.
|
|
"""
|
|
tool_def = ToolDefinition(name='weather', description='get the weather before\udce4after')
|
|
|
|
model = InstrumentedModel(MyModel())
|
|
messages: list[ModelMessage] = [ModelRequest(parts=[UserPromptPart('Hello')], timestamp=IsDatetime())]
|
|
await model.request(
|
|
messages,
|
|
model_settings=None,
|
|
model_request_parameters=ModelRequestParameters(function_tools=[tool_def]),
|
|
)
|
|
|
|
attrs = capfire.exporter.exported_spans_as_dict(parse_json_attributes=True)[0]['attributes']
|
|
assert attrs['gen_ai.tool.definitions'] == snapshot(
|
|
[
|
|
{
|
|
'type': 'function',
|
|
'name': 'weather',
|
|
'description': 'get the weather before\udce4after',
|
|
'parameters': {'type': 'object', 'properties': {}},
|
|
}
|
|
]
|
|
)
|
|
assert 'weather' in attrs['model_request_parameters']['function_tools'][0]['name']
|
|
|
|
|
|
async def test_instrumented_model_request_error(capfire: CaptureLogfire):
|
|
"""Test _instrument() when the wrapped model raises before finish() is called."""
|
|
|
|
class ErrorModel(MyModel):
|
|
async def request(
|
|
self,
|
|
messages: list[ModelMessage],
|
|
model_settings: ModelSettings | None,
|
|
model_request_parameters: ModelRequestParameters,
|
|
) -> ModelResponse:
|
|
raise RuntimeError('model error')
|
|
|
|
model = InstrumentedModel(ErrorModel())
|
|
messages: list[ModelMessage] = [ModelRequest(parts=[UserPromptPart('Hello')], timestamp=IsDatetime())]
|
|
|
|
with pytest.raises(RuntimeError, match='model error'):
|
|
await model.request(
|
|
messages,
|
|
model_settings=None,
|
|
model_request_parameters=ModelRequestParameters(),
|
|
)
|
|
|
|
# Span should still be created, but without finish()-specific attributes
|
|
spans = capfire.exporter.exported_spans_as_dict(parse_json_attributes=True)
|
|
assert len(spans) == 1
|
|
assert spans[0]['attributes']['gen_ai.request.model'] == 'gpt-4o'
|
|
# finish() was never called, so response-specific attributes are absent
|
|
assert 'gen_ai.response.id' not in spans[0]['attributes']
|
|
assert 'gen_ai.usage.input_tokens' not in spans[0]['attributes']
|