`symfonic.core.callbacks.emit`¶

emit ¶

Helpers to emit LLM callback events from non-react call sites (v7.4.3).

Centralises the "mirror react.py:303-311" pattern so the seven non-react LLM call sites (metacognition critic, context-window summariser, insight extractor, LLM entity extractor, consolidation extractor, procedural router, plus future siblings) all emit identical LLMEndEvent payloads with the right node_name discriminator.

Design rules:

Always safe with callback_manager=None (just returns early).
Always safe with the manager's is_noop zero-cost path.
Never raises -- any exception in a handler is swallowed by the manager's own dispatcher; this helper does not introduce a second layer of guards.
Constructs the event eagerly only when there's at least one handler registered -- mirrors the react.py guard at react.py:303.

Used by:

agent/middleware/metacognitive.py (metacognition_critic)
agent/engine.py (metacognition_skill_gate via middleware)
core/nodes/context_window.py (context_window_summariser)
core/learning/refiner.py (insight_extractor)
core/learning/entity_extractor_llm.py (entity_extractor_llm)
memory/_internal/extraction/service.py (consolidation_extractor)
memory/layers/procedural/router.py (procedural_router)

LLMTiming `dataclass` ¶

LLMTiming(
    started_at_utc: str = "", duration_ms: float = 0.0
)

Capture window for LLM call timing (v7.15.0).

Populated by the :func:llm_timing async context manager. Passed into :func:emit_llm_end / :func:emit_llm_end_from_result via the timing kwarg so the resulting :class:LLMEndEvent carries duration_ms and started_at_utc for adopter cost-over-time aggregation and tail-latency analysis.

Attributes:

Name	Type	Description
`started_at_utc`	`str`	ISO-8601 UTC timestamp of the LLM invocation start. Empty string when the timing context never opened.
`duration_ms`	`float`	Wall-clock duration in milliseconds. `0.0` when the timing context never closed (e.g. handler dropped the timing reference before exiting the `async with`).

emit_llm_end `async` ¶

emit_llm_end(
    callback_manager: CallbackManager | None,
    *,
    model: str,
    output: str,
    usage: dict[str, Any],
    run_id: str,
    node_name: str,
    timing: LLMTiming | None = None,
    iteration_index: int = 0,
) -> None

Fire on_llm_end on the manager, mirroring react.py:303-311.

No-ops when the manager is None or registers zero handlers. Never raises. The event is constructed eagerly only when a handler is present so the empty-manager fast path stays zero-cost.

Parameters:

Name	Type	Description	Default
`callback_manager`	`CallbackManager \| None`	Merged manager from runtime, or `None` for callers that opted out. `None` is the default for every seven bypass-site signature so back-compat is preserved.	required
`model`	`str`	`ModelConfig.model_name` (or whatever string the caller used in its `ainvoke`). Anthropic billing buckets by model.	required
`output`	`str`	Stringified LLM output. Use `_extract_output_text` to handle the AIMessage / raw-string union.	required
`usage`	`dict[str, Any]`	Provider usage dict (`input_tokens` / `output_tokens`). Use `_extract_usage` for the standard extraction.	required
`run_id`	`str`	Engine run identifier so OTel and metrics can correlate with the surrounding run. Empty string is acceptable for ad-hoc consolidation-phase calls.	required
`node_name`	`str`	Discriminator string -- one of the documented set in the module docstring. Required (no default) so callers think about it.	required
`timing`	`LLMTiming \| None`	v7.15.0 :class:`LLMTiming` captured via the :func:`llm_timing` context manager. `None` (default) produces `duration_ms=0.0`, `started_at_utc=""` -- byte-identical pre-v7.15.0 `LLMEndEvent`.	`None`
`iteration_index`	`int`	v7.15.0 0-indexed react loop iteration that produced this LLM call. Defaults to `0` for non-react sites (single emission per consolidation cycle).	`0`

Source code in src/symfonic/core/callbacks/emit.py

async def emit_llm_end(
    callback_manager: CallbackManager | None,
    *,
    model: str,
    output: str,
    usage: dict[str, Any],
    run_id: str,
    node_name: str,
    timing: LLMTiming | None = None,
    iteration_index: int = 0,
) -> None:
    """Fire ``on_llm_end`` on the manager, mirroring react.py:303-311.

    No-ops when the manager is ``None`` or registers zero handlers.  Never
    raises.  The event is constructed eagerly only when a handler is present
    so the empty-manager fast path stays zero-cost.

    Args:
        callback_manager: Merged manager from runtime, or ``None`` for
            callers that opted out.  ``None`` is the default for every
            seven bypass-site signature so back-compat is preserved.
        model: ``ModelConfig.model_name`` (or whatever string the caller
            used in its ``ainvoke``).  Anthropic billing buckets by model.
        output: Stringified LLM output.  Use ``_extract_output_text`` to
            handle the AIMessage / raw-string union.
        usage: Provider usage dict (``input_tokens`` / ``output_tokens``).
            Use ``_extract_usage`` for the standard extraction.
        run_id: Engine run identifier so OTel and metrics can correlate
            with the surrounding run.  Empty string is acceptable for
            ad-hoc consolidation-phase calls.
        node_name: Discriminator string -- one of the documented set in
            the module docstring.  Required (no default) so callers think
            about it.
        timing: v7.15.0 :class:`LLMTiming` captured via the
            :func:`llm_timing` context manager.  ``None`` (default)
            produces ``duration_ms=0.0``, ``started_at_utc=""`` --
            byte-identical pre-v7.15.0 ``LLMEndEvent``.
        iteration_index: v7.15.0 0-indexed react loop iteration that
            produced this LLM call.  Defaults to ``0`` for non-react
            sites (single emission per consolidation cycle).
    """
    if callback_manager is None or callback_manager.is_noop:
        return

    from ..contracts.callbacks import LLMEndEvent

    event = LLMEndEvent(
        model=model,
        output=output,
        usage=usage,
        run_id=run_id,
        node_name=node_name,
        duration_ms=timing.duration_ms if timing else 0.0,
        started_at_utc=timing.started_at_utc if timing else "",
        iteration_index=iteration_index,
    )
    await callback_manager.on_llm_end(event)

emit_llm_end_from_result `async` ¶

emit_llm_end_from_result(
    callback_manager: CallbackManager | None,
    *,
    model: str,
    result: Any,
    run_id: str,
    node_name: str,
    timing: LLMTiming | None = None,
    iteration_index: int = 0,
) -> None

Convenience wrapper: extract usage/output from a model result and emit.

Equivalent to::

await emit_llm_end(
    cb_mgr,
    model=model,
    output=_extract_output_text(result),
    usage=_extract_usage(result),
    run_id=run_id,
    node_name=node_name,
    timing=timing,
    iteration_index=iteration_index,
)

Use this at sites where the caller already has the raw ainvoke result in hand (most of the seven bypass sites).

v7.15.0 added timing and iteration_index -- both default to None / 0 so existing callers see no behaviour change. Callers that want the new telemetry wrap their ainvoke with :func:llm_timing and pass the captured :class:LLMTiming here.

Source code in src/symfonic/core/callbacks/emit.py

async def emit_llm_end_from_result(
    callback_manager: CallbackManager | None,
    *,
    model: str,
    result: Any,
    run_id: str,
    node_name: str,
    timing: LLMTiming | None = None,
    iteration_index: int = 0,
) -> None:
    """Convenience wrapper: extract usage/output from a model result and emit.

    Equivalent to::

        await emit_llm_end(
            cb_mgr,
            model=model,
            output=_extract_output_text(result),
            usage=_extract_usage(result),
            run_id=run_id,
            node_name=node_name,
            timing=timing,
            iteration_index=iteration_index,
        )

    Use this at sites where the caller already has the raw ``ainvoke``
    result in hand (most of the seven bypass sites).

    v7.15.0 added ``timing`` and ``iteration_index`` -- both default to
    None / 0 so existing callers see no behaviour change.  Callers that
    want the new telemetry wrap their ``ainvoke`` with
    :func:`llm_timing` and pass the captured :class:`LLMTiming` here.
    """
    if callback_manager is None or callback_manager.is_noop:
        return
    await emit_llm_end(
        callback_manager,
        model=model,
        output=_extract_output_text(result),
        usage=_extract_usage(result),
        run_id=run_id,
        node_name=node_name,
        timing=timing,
        iteration_index=iteration_index,
    )

emit_llm_start `async` ¶

emit_llm_start(
    callback_manager: CallbackManager | None,
    *,
    model: str,
    messages: list,
    run_id: str,
    node_name: str,
    system_prompt: str = "",
) -> None

Fire on_llm_start on the manager, mirroring react.py:1082-1090.

THE Jarvio fix (v8.4.0): the eight aux LLM call sites (critic, summariser, insight extractor, consolidation extractor, procedural router, entity/procedural extractors, synthesis fallback) historically emitted only the callback LLMEndEvent -- never the paired LLMStartEvent. The OTel CallbackBridge OPENS a span on on_llm_start and CLOSES it on on_llm_end; with no start the end_llm pops a None key and returns silently, so NO span is ever recorded in cortex. React was the only site emitting the start, which is why only react spans appeared.

Calling this helper BEFORE the aux ainvoke (with the SAME node_name discriminator the site already passes to :func:emit_llm_end, on the SAME merged manager) opens the span so the existing end closes it. Carries node_name on the :class:LLMStartEvent (v8.4.0 added the field) so the bridge can key the span per node -- concurrent-safe attribution within a single turn.

No-ops when the manager is None or registers zero handlers (is_noop). Never raises (the manager's dispatcher swallows handler exceptions). Byte-identical to pre-v8.4.0 behaviour when no callback manager is merged.

Parameters:

Name	Type	Description	Default
`callback_manager`	`CallbackManager \| None`	Merged manager from runtime, or `None` for callers that opted out. `None` / noop is the byte-identical fast path.	required
`model`	`str`	`ModelConfig.model_name` or the resolved SKU string the caller used in its `ainvoke` (typically the same value passed to :func:`emit_llm_end`).	required
`messages`	`list`	The conversation-history message list passed to `ainvoke` (engine-view, pre tool-binding). Empty list is acceptable for sites that build a single prompt string.	required
`run_id`	`str`	Engine turn identifier so OTel can correlate with the surrounding run. Empty string is acceptable for ad-hoc consolidation-phase calls.	required
`node_name`	`str`	Discriminator string -- one of the documented set in the module docstring. Required (no default) so callers think about it; MUST match the `node_name` the paired :func:`emit_llm_end` uses.	required
`system_prompt`	`str`	Assembled system prompt string, surfaced to CallbackHandlers (e.g. prompt capture). Empty when the site uses no separate system prompt.	`''`

Source code in src/symfonic/core/callbacks/emit.py

async def emit_llm_start(
    callback_manager: CallbackManager | None,
    *,
    model: str,
    messages: list,
    run_id: str,
    node_name: str,
    system_prompt: str = "",
) -> None:
    """Fire ``on_llm_start`` on the manager, mirroring ``react.py:1082-1090``.

    THE Jarvio fix (v8.4.0): the eight aux LLM call sites (critic,
    summariser, insight extractor, consolidation extractor, procedural
    router, entity/procedural extractors, synthesis fallback) historically
    emitted only the callback ``LLMEndEvent`` -- never the paired
    ``LLMStartEvent``.  The OTel ``CallbackBridge`` OPENS a span on
    ``on_llm_start`` and CLOSES it on ``on_llm_end``; with no start the
    ``end_llm`` pops a ``None`` key and returns silently, so NO span is
    ever recorded in cortex.  React was the only site emitting the start,
    which is why only react spans appeared.

    Calling this helper BEFORE the aux ``ainvoke`` (with the SAME
    ``node_name`` discriminator the site already passes to
    :func:`emit_llm_end`, on the SAME merged manager) opens the span so the
    existing end closes it.  Carries ``node_name`` on the
    :class:`LLMStartEvent` (v8.4.0 added the field) so the bridge can key
    the span per node -- concurrent-safe attribution within a single turn.

    No-ops when the manager is ``None`` or registers zero handlers
    (``is_noop``).  Never raises (the manager's dispatcher swallows handler
    exceptions).  Byte-identical to pre-v8.4.0 behaviour when no callback
    manager is merged.

    Args:
        callback_manager: Merged manager from runtime, or ``None`` for
            callers that opted out.  ``None`` / noop is the byte-identical
            fast path.
        model: ``ModelConfig.model_name`` or the resolved SKU string the
            caller used in its ``ainvoke`` (typically the same value passed
            to :func:`emit_llm_end`).
        messages: The conversation-history message list passed to
            ``ainvoke`` (engine-view, pre tool-binding).  Empty list is
            acceptable for sites that build a single prompt string.
        run_id: Engine turn identifier so OTel can correlate with the
            surrounding run.  Empty string is acceptable for ad-hoc
            consolidation-phase calls.
        node_name: Discriminator string -- one of the documented set in
            the module docstring.  Required (no default) so callers think
            about it; MUST match the ``node_name`` the paired
            :func:`emit_llm_end` uses.
        system_prompt: Assembled system prompt string, surfaced to
            CallbackHandlers (e.g. prompt capture).  Empty when the site
            uses no separate system prompt.
    """
    if callback_manager is None or callback_manager.is_noop:
        return

    from ..contracts.callbacks import LLMStartEvent

    event = LLMStartEvent(
        model=model,
        messages=messages,
        run_id=run_id,
        system_prompt=system_prompt,
        node_name=node_name,
    )
    await callback_manager.on_llm_start(event)

llm_timing `async` ¶

llm_timing() -> AsyncIterator[LLMTiming]

Wrap an LLM call to capture wall-clock duration and start timestamp.

Usage::

async with llm_timing() as timing:
    result = await model.ainvoke(...)
await emit_llm_end_from_result(..., timing=timing)

The context manager is exception-safe: duration_ms is populated even when ainvoke raises. Callers that don't care about timing can omit the timing kwarg from the emit helpers -- the defaults preserve the pre-v7.15.0 contract (duration_ms=0.0, started_at_utc="") so adopter handlers that ignore the new fields see byte-identical behaviour.

Yields:

Type	Description
`AsyncIterator[LLMTiming]`	class:`LLMTiming` -- populated with `started_at_utc` on
`AsyncIterator[LLMTiming]`	entry, then `duration_ms` on exit. Safe to read after the
`AsyncIterator[LLMTiming]`	context exits.

Source code in src/symfonic/core/callbacks/emit.py

@asynccontextmanager
async def llm_timing() -> AsyncIterator[LLMTiming]:
    """Wrap an LLM call to capture wall-clock duration and start timestamp.

    Usage::

        async with llm_timing() as timing:
            result = await model.ainvoke(...)
        await emit_llm_end_from_result(..., timing=timing)

    The context manager is exception-safe: ``duration_ms`` is populated
    even when ``ainvoke`` raises.  Callers that don't care about timing
    can omit the ``timing`` kwarg from the emit helpers -- the defaults
    preserve the pre-v7.15.0 contract (``duration_ms=0.0``,
    ``started_at_utc=""``) so adopter handlers that ignore the new
    fields see byte-identical behaviour.

    Yields:
        :class:`LLMTiming` -- populated with ``started_at_utc`` on
        entry, then ``duration_ms`` on exit.  Safe to read after the
        context exits.
    """
    timing = LLMTiming(
        started_at_utc=datetime.now(UTC).isoformat(),
    )
    t0 = monotonic()
    try:
        yield timing
    finally:
        timing.duration_ms = (monotonic() - t0) * 1000.0

resolve_model_name ¶

resolve_model_name(model: Any, fallback: str = '') -> str

Extract the actual-API-call model identifier from a BaseChatModel.

The framework historically stamped LLMEndEvent.model with the ModelConfig.model_name requested at the call site, but adopters that route via MultiProviderRouter or a custom ModelProvider (Jarvio's JarvioModelProvider) can return a chat model bound to a different SKU than was requested. Reporting the requested-name under-attributes cost to the wrong pricing row (the v7.13.6 fix revealed this -- Jarvio's stack requested claude-sonnet-4-... via the framework default but the provider returned a Claude Opus 4.6 chat model, which costs ~3x more).

Provider-attribute survey (2026-06-01):

============================================ ============ ========== Concrete class .model .model_name ============================================ ============ ========== langchain_anthropic.ChatAnthropic str None langchain_openai.ChatOpenAI None str langchain_google_genai.ChatGoogleGenerativeAI str None langchain_ollama.ChatOllama str None symfonic.core.testing.MockChatModel absent absent langchain_core.runnables.RunnableBinding absent absent (LangChain's bind_tools wrapper) ============================================ ============ ==========

Strategy: try .model first, then .model_name, then descend once through .bound (the RunnableBinding wrapper exposed by BaseChatModel.bind_tools). Empty strings are treated as "missing" so callers don't end up stamping "" on the cost row. The fallback argument carries the previous behaviour (the ModelConfig.model_name string) so any unknown wrapper degrades to the pre-v7.14.0 contract instead of crashing.

Parameters:

Name	Type	Description	Default
`model`	`Any`	The `BaseChatModel` or `RunnableBinding` actually invoked. `None` is accepted and routes to `fallback`.	required
`fallback`	`str`	String returned when none of the candidate attributes yield a non-empty string -- typically the requested `ModelConfig.model_name`.	`''`

Returns:

Type	Description
`str`	The resolved model identifier (e.g. `"claude-opus-4-6"`) when
`str`	available, otherwise `fallback`.

Source code in src/symfonic/core/callbacks/emit.py

def resolve_model_name(model: Any, fallback: str = "") -> str:
    """Extract the actual-API-call model identifier from a ``BaseChatModel``.

    The framework historically stamped ``LLMEndEvent.model`` with the
    ``ModelConfig.model_name`` requested at the call site, but adopters
    that route via ``MultiProviderRouter`` or a custom ``ModelProvider``
    (Jarvio's ``JarvioModelProvider``) can return a chat model bound to
    a different SKU than was requested.  Reporting the requested-name
    under-attributes cost to the wrong pricing row (the v7.13.6 fix
    revealed this -- Jarvio's stack requested ``claude-sonnet-4-...``
    via the framework default but the provider returned a Claude Opus
    4.6 chat model, which costs ~3x more).

    Provider-attribute survey (2026-06-01):

    ============================================  ============  ==========
    Concrete class                                 ``.model``    ``.model_name``
    ============================================  ============  ==========
    ``langchain_anthropic.ChatAnthropic``          str           None
    ``langchain_openai.ChatOpenAI``                None          str
    ``langchain_google_genai.ChatGoogleGenerativeAI`` str        None
    ``langchain_ollama.ChatOllama``                str           None
    ``symfonic.core.testing.MockChatModel``        absent        absent
    ``langchain_core.runnables.RunnableBinding``   absent        absent
        (LangChain's ``bind_tools`` wrapper)
    ============================================  ============  ==========

    Strategy: try ``.model`` first, then ``.model_name``, then descend
    once through ``.bound`` (the RunnableBinding wrapper exposed by
    ``BaseChatModel.bind_tools``).  Empty strings are treated as
    "missing" so callers don't end up stamping ``""`` on the cost row.
    The ``fallback`` argument carries the previous behaviour (the
    ``ModelConfig.model_name`` string) so any unknown wrapper degrades
    to the pre-v7.14.0 contract instead of crashing.

    Args:
        model: The ``BaseChatModel`` or ``RunnableBinding`` actually
            invoked.  ``None`` is accepted and routes to ``fallback``.
        fallback: String returned when none of the candidate attributes
            yield a non-empty string -- typically the requested
            ``ModelConfig.model_name``.

    Returns:
        The resolved model identifier (e.g. ``"claude-opus-4-6"``) when
        available, otherwise ``fallback``.
    """
    if model is None:
        return fallback
    for attr in ("model", "model_name"):
        v = getattr(model, attr, None)
        if isinstance(v, str) and v:
            return v
    # RunnableBinding (bind_tools wrapper) exposes the underlying chat
    # model under ``.bound``.  One level of descent is enough -- bind_tools
    # never returns a doubly-wrapped binding in any langchain version we
    # support.
    inner = getattr(model, "bound", None)
    if inner is not None and inner is not model:
        for attr in ("model", "model_name"):
            v = getattr(inner, attr, None)
            if isinstance(v, str) and v:
                return v
    return fallback

symfonic.core.callbacks.emit¶

emit ¶

LLMTiming dataclass ¶

emit_llm_end async ¶

emit_llm_end_from_result async ¶

emit_llm_start async ¶

llm_timing async ¶

resolve_model_name ¶

`symfonic.core.callbacks.emit`¶

LLMTiming `dataclass` ¶

emit_llm_end `async` ¶

emit_llm_end_from_result `async` ¶

emit_llm_start `async` ¶

llm_timing `async` ¶