Skip to content

symfonic.agent.backend.conversation

conversation

The conversation and model ports, in LangChain's vocabulary.

Kernel values and provider chunks meet here; the kernel stays stdlib-pure.

ConversationAdapter 

ConversationAdapter(provider: Any, *, history_cache: Any = None)

Bases: ReceiptSelectionAdapter

Builds and records completed conversation rounds.

Source code in src/symfonic/agent/backend/conversation.py 
45
46
47
def __init__(self, provider: Any, *, history_cache: Any = None) -> None:
    self._provider = provider
    self._history_cache = history_cache

close_round 

close_round(transcript: Transcript, turn: ModelTurn, requests: Sequence[ToolRequest], outcomes: Sequence[Any]) -> None

Append a round's reply and tool observations. Terminal extraction reads the answer from wire (RES-8). Receipts bind turn.payload by identity; only the wire copy is converted.

Source code in src/symfonic/agent/backend/conversation.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def close_round(
    self,
    transcript: Transcript,
    turn: ModelTurn,
    requests: Sequence[ToolRequest],
    outcomes: Sequence[Any],
) -> None:
    """Append a round's reply and tool observations.
    Terminal extraction reads the answer from wire (RES-8). Receipts
    bind ``turn.payload`` by identity; only the wire copy is converted.
    """
    transcript._select_reply(turn)
    transcript.typed.append(assistant_message(turn.text, tuple(outcomes), turn.payload))
    if not outcomes:
        transcript.append_wire(replayed_reply(turn.payload), preserve_view=True)
        return

    transcript._latest_round_start = len(transcript.wire)
    reply = replayed_reply(turn.payload, [request.call_id for request in requests])
    transcript.append_wire(reply)
    for outcome, observation in zip(outcomes, tool_observations(reply, outcomes), strict=True):
        transcript.typed.append(
            Message(
                role="tool",
                content=observation.content,
                tool_call_id=outcome.id,
                tool_status="error" if outcome.error is not None else None,
            )
        )
        transcript.append_wire(observation)

ModelAdapter 

ModelAdapter(bound: Any, *, observe_provider_call: Callable[[object], None] | None = None)

Executes the plan's model decision; never re-picks one (ModelPort).

Source code in src/symfonic/agent/backend/conversation.py 
118
119
120
121
122
def __init__(
    self, bound: Any, *, observe_provider_call: Callable[[object], None] | None = None
) -> None:
    self._bound = bound
    self._observe_provider_call = observe_provider_call

StreamingRound 

StreamingRound(bound: Any, transcript: Transcript, observe: Callable[[object], None] | None = None)

One streaming round, aggregating chunks as it yields deltas.

Source code in src/symfonic/agent/backend/conversation.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
def __init__(
    self,
    bound: Any,
    transcript: Transcript,
    observe: Callable[[object], None] | None = None,
) -> None:
    self._bound = bound
    self._transcript = transcript
    self._observe = observe
    self._aggregate: Any = None
    self._filter = ExtractionFilter()
    self._text: list[str] = []
    self._done = False

result 

result() -> ModelTurn

Return the completed round using exactly the emitted deltas.

Source code in src/symfonic/agent/backend/conversation.py 
209
210
211
212
213
214
215
216
217
218
219
def result(self) -> ModelTurn:
    """Return the completed round using exactly the emitted deltas."""
    record = (self._transcript._record_successful_reply if self._done
              else self._transcript._record_reply)
    turn = record(self._aggregate, turn_of(
        self._aggregate,
        "".join(self._text),
        extraction_ops=tuple(self._filter.get_extracted_ops()),
        extraction_seen=self._filter.extraction_seen,
    ))
    return turn

turn_of 

turn_of(response: Any, text: str, *, extraction_ops: tuple[dict[str, Any], ...] = (), extraction_seen: bool = False) -> ModelTurn

Project one provider message onto the kernel's ModelTurn.

Source code in src/symfonic/agent/backend/conversation.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def turn_of(
    response: Any,
    text: str,
    *,
    extraction_ops: tuple[dict[str, Any], ...] = (),
    extraction_seen: bool = False,
) -> ModelTurn:
    """Project one provider message onto the kernel's ``ModelTurn``."""
    extraction = ExtractionFilter()
    clean_text = extraction.feed(text) + extraction.flush()
    if not extraction_seen:
        extraction_seen = extraction.extraction_seen
        extraction_ops = tuple(extraction.get_extracted_ops())
    requested = list(getattr(response, "tool_calls", None) or [])
    return ModelTurn(
        text=clean_text,
        tool_requests=tuple(
            ToolRequest(
                call_id=call.get("id") or "",
                name=call.get("name", ""),
                arguments=dict(call.get("args") or {}),
            )
            for call in requested
        ),
        usage=_usage(response),
        stop_reason=normalize_stop_reason(response),
        payload=response,
        extraction_ops=extraction_ops,
        extraction_seen=extraction_seen,
    )