symfonic.agent.backend.messages

messages

Conversion between the facade's typed Message and LangChain messages.

One direction builds the request (RES-2's "system, history, user" ordering); the other records what came back. Both are pure functions so the invocation loop stays readable and this file is the only place the wire shape is known.

assistant_message

assistant_message(text: str, calls: Sequence[ToolCall]) -> Message

The facade-typed record of one assistant turn.

The same ToolCall objects appear here and on AgentResult.tool_calls, so a caller may walk either view and compare by identity (RES-3).

Source code in src/symfonic/agent/backend/messages.py

def assistant_message(text: str, calls: Sequence[ToolCall]) -> Message:
    """The facade-typed record of one assistant turn.

    The same ``ToolCall`` objects appear here and on ``AgentResult.tool_calls``,
    so a caller may walk either view and compare by identity (RES-3).
    """
    return Message(role="assistant", content=text, tool_calls=tuple(calls))

build_request

build_request(prompt: str, *, instructions: str | None, system_blocks: Sequence[Any] = (), history: Sequence[Message], attachments: Sequence[Attachment], family: ProviderFamily) -> tuple[list[Message], list[BaseMessage]]

Return the turn's facade messages and their LangChain equivalents.

Order is RES-2's: the system message when instructions is set, the replayed history, then the user message. instructions is used verbatim — None sends no system message rather than substituting a framework default (FAC-5).

When instructions is set, system messages already in history are dropped rather than replayed. RES-2 makes result.messages a lossless round trip, and those messages carry the previous turn's system message — replaying it alongside the freshly prepended one sends the instructions twice on turn two, three times on turn three, since providers concatenate system blocks instead of deduplicating them. instructions is the one source of the system prompt for this agent, so the replayed copies go. When instructions is None the history is replayed untouched: a transcript handed to an agent that declares no instructions of its own keeps the ones it was recorded with.

Source code in src/symfonic/agent/backend/messages.py

def build_request(
    prompt: str,
    *,
    instructions: str | None,
    system_blocks: Sequence[Any] = (),
    history: Sequence[Message],
    attachments: Sequence[Attachment],
    family: ProviderFamily,
) -> tuple[list[Message], list[BaseMessage]]:
    """Return the turn's facade messages and their LangChain equivalents.

    Order is RES-2's: the system message when ``instructions`` is set, the
    replayed history, then the user message. ``instructions`` is used verbatim
    — ``None`` sends no system message rather than substituting a framework
    default (FAC-5).

    When ``instructions`` is set, system messages already in ``history`` are
    dropped rather than replayed. RES-2 makes ``result.messages`` a lossless
    round trip, and those messages *carry the previous turn's system message*
    — replaying it alongside the freshly prepended one sends the instructions
    twice on turn two, three times on turn three, since providers concatenate
    system blocks instead of deduplicating them. ``instructions`` is the one
    source of the system prompt for this agent, so the replayed copies go.
    When ``instructions`` is ``None`` the history is replayed untouched: a
    transcript handed to an agent that declares no instructions of its own
    keeps the ones it was recorded with.
    """
    typed: list[Message] = []
    if instructions is not None:
        typed.append(Message(role="system", content=instructions))
        typed.extend(m for m in history if m.role != "system")
    else:
        typed.extend(history)
    typed.append(Message(role="user", content=prompt))

    wire: list[BaseMessage] = [to_langchain(m) for m in typed]
    # Anthropic's LangChain serializer recognises its cache control only on a
    # structured system-message content list.  Other providers keep the
    # portable text projection; a prompt cache is not a cross-provider history
    # cache policy.
    if (
        system_blocks
        and family == "anthropic"
        and instructions is not None
        and _blocks_text(system_blocks) == instructions
    ):
        wire[0] = SystemMessage(content=list(system_blocks))
    if attachments:
        # The user message is always last; give it the multimodal content list.
        wire[-1] = HumanMessage(
            content=_build_human_content(prompt, list(attachments), family)
        )
    return typed, wire

normalize_stop_reason

normalize_stop_reason(message: BaseMessage | Any) -> str | None

Map the provider's terminal reason onto the facade's four, or None.

Source code in src/symfonic/agent/backend/messages.py

def normalize_stop_reason(message: BaseMessage | Any) -> str | None:
    """Map the provider's terminal reason onto the facade's four, or ``None``."""
    metadata = getattr(message, "response_metadata", None) or {}
    raw = metadata.get("stop_reason") or metadata.get("finish_reason")
    if not isinstance(raw, str):
        return None
    return _STOP_REASONS.get(raw.lower())

text_of

text_of(message: BaseMessage | Any) -> str

Flatten LangChain message content to text; never None (RES-1).

Content is a str for text-only replies and a list of typed blocks when the provider interleaves reasoning or tool use; only text blocks are answer text.

Source code in src/symfonic/agent/backend/messages.py

def text_of(message: BaseMessage | Any) -> str:
    """Flatten LangChain message content to text; never ``None`` (RES-1).

    Content is a ``str`` for text-only replies and a list of typed blocks when
    the provider interleaves reasoning or tool use; only ``text`` blocks are
    answer text.
    """
    content = getattr(message, "content", "")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return "".join(
            part.get("text", "")
            for part in content
            if isinstance(part, dict) and part.get("type") == "text"
        )
    return ""

to_langchain

to_langchain(message: Message) -> BaseMessage

Convert one facade message into its LangChain equivalent.

content is the str Message declares, except for a message replayed out of caller history that carried a LangChain block list — an image, a document, a provider reasoning block. Message.content has no room for those, so the converter that built the message kept them on the side (blocks) and this is where they go back on the wire. Dropping them here would hand the provider a conversation with the picture removed while the legacy route, which replays the caller's BaseMessage verbatim, hands it the picture.

Source code in src/symfonic/agent/backend/messages.py

def to_langchain(message: Message) -> BaseMessage:
    """Convert one facade message into its LangChain equivalent.

    ``content`` is the ``str`` ``Message`` declares, except for a message
    replayed out of caller ``history`` that carried a LangChain *block* list —
    an image, a document, a provider reasoning block. ``Message.content`` has
    no room for those, so the converter that built the message kept them on
    the side (``blocks``) and this is where they go back on the wire. Dropping
    them here would hand the provider a conversation with the picture removed
    while the legacy route, which replays the caller's ``BaseMessage``
    verbatim, hands it the picture.
    """
    content = _content_of(message)
    if message.role == "system":
        return SystemMessage(content=content)
    if message.role == "user":
        return HumanMessage(content=content)
    if message.role == "tool":
        return ToolMessage(content=content, tool_call_id=message.tool_call_id or "")
    return AIMessage(
        content=content,
        tool_calls=[
            {"name": c.name, "args": dict(c.arguments), "id": c.id}
            for c in message.tool_calls
        ],
    )