A span per model call in the OpenTelemetry GenAI semantic conventions (#680).
Observability tools such as Langfuse, Opik and Phoenix recognise a model call by its gen_ai.* attributes; Flock's own run totals (flock.usage.*, #571) are not among them. :class:GenAISpans is a DSPy callback that opens a LM.chat span around every LM call an engine makes, under the span that is current at the call, and records the model, the provider, the tokens and the Flock agent.
Message content is recorded only with OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true, each side capped by FLOCK_TRACE_MAX_ATTRIBUTE_BYTES. FLOCK_TRACE_IGNORE='["LM.chat"]' switches the spans off.
Classes
GenAISpans(agent_name: str | None = None, correlation_id: object = None)
Bases: BaseCallback
DSPy callback: one LM.chat span per LM call, for one agent.
Source code in src/flock/logging/genai_spans.py
| def __init__(
self, agent_name: str | None = None, correlation_id: object = None
) -> None:
self.agent_name = agent_name
# The agent run's correlation, so model calls join its session (#681)
self.correlation_id = correlation_id
self._calls: dict[str, tuple[Any, Any]] = {}
|
Functions:
provider_name(model: str) -> str
gen_ai.provider.name for a LiteLLM model name.
Source code in src/flock/logging/genai_spans.py
| def provider_name(model: str) -> str:
"""``gen_ai.provider.name`` for a LiteLLM model name."""
try:
import litellm
provider = litellm.get_llm_provider(model)[1]
except Exception: # noqa: BLE001 - LiteLLM raises for models it does not map
provider = model.split("/", 1)[0] if "/" in model else "openai"
return _PROVIDERS.get(provider, provider)
|
capture_content() -> bool
Whether message content goes on the spans (off by default, as the conventions say).
Source code in src/flock/logging/genai_spans.py
| def capture_content() -> bool:
"""Whether message content goes on the spans (off by default, as the conventions say)."""
value = os.getenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "")
return value.strip().lower() in {"true", "1", "yes", "on"}
|
A request as the conventions' input messages: a role and parts each.
Source code in src/flock/logging/genai_spans.py
| def input_messages(messages: Any, prompt: Any = None) -> list[dict[str, Any]]:
"""A request as the conventions' input messages: a role and parts each."""
if not messages:
return [{"role": "user", "parts": _parts(prompt)}]
result = []
for message in messages:
role = message.get("role", "user")
if role == "tool":
parts = [
{
"type": "tool_call_response",
"id": message.get("tool_call_id"),
"response": message.get("content"),
}
]
else:
parts = _parts(message.get("content")) + [
_tool_call(call) for call in message.get("tool_calls") or []
]
result.append({"role": role, "parts": parts})
return result
|
DSPy's outputs as the conventions' output messages; finish_reason when the response names one for that choice.
Source code in src/flock/logging/genai_spans.py
| def output_messages(outputs: Any, response: Any = None) -> list[dict[str, Any]]:
"""DSPy's outputs as the conventions' output messages; ``finish_reason`` when
the response names one for that choice."""
reasons = [
getattr(choice, "finish_reason", None)
for choice in getattr(response, "choices", None) or []
]
result = []
for index, output in enumerate(outputs or []):
if isinstance(output, dict):
parts = []
if output.get("reasoning_content"):
parts.append({
"type": "reasoning",
"content": output["reasoning_content"],
})
if output.get("text") is not None:
parts.append({"type": "text", "content": output["text"]})
parts += [_tool_call(call) for call in output.get("tool_calls") or []]
else:
parts = _parts(output)
message: dict[str, Any] = {"role": "assistant", "parts": parts}
if index < len(reasons) and reasons[index]:
message["finish_reason"] = str(reasons[index])
result.append(message)
return result
|