Skip to content

symfonic.services.conversation.history

history

History strategies, separated from agent orchestration.

The legacy strategies resolved by mutating an AgentConfig, so reasoning about "what will this conversation keep?" meant reading the engine. Here a strategy answers one question — :class:HistoryDirective — and translating that answer into legacy config overrides is a separate, adapter-side step (to_legacy_overrides). This module imports nothing from the engine.

HistoryDirective dataclass 

HistoryDirective(kind: HistoryKind, summarize: bool = False, window_messages: int | None = None, trigger_chars: int | None = None, keep_recent: int | None = None)

What a strategy decided, as an inert value.

default classmethod 

default() -> HistoryDirective

The framework's documented default: summarize overflow.

Source code in src/symfonic/services/conversation/history.py 
50
51
52
53
54
55
56
57
58
@classmethod
def default(cls) -> HistoryDirective:
    """The framework's documented default: summarize overflow."""
    return cls(
        kind="summarizing",
        summarize=True,
        trigger_chars=_DEFAULT_TRIGGER_CHARS,
        keep_recent=_DEFAULT_KEEP_RECENT,
    )

to_legacy_overrides 

to_legacy_overrides() -> dict[str, Any]

Render as the legacy config fields, carrying no engine types.

Both mechanisms are always named. Leaving one unset would let whatever the adopter's config already held govern alongside this directive, which is precisely the "two limiters, unclear winner" ambiguity the strategy objects exist to remove.

Source code in src/symfonic/services/conversation/history.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def to_legacy_overrides(self) -> dict[str, Any]:
    """Render as the legacy config fields, carrying no engine types.

    Both mechanisms are always named. Leaving one unset would let whatever
    the adopter's config already held govern alongside this directive,
    which is precisely the "two limiters, unclear winner" ambiguity the
    strategy objects exist to remove.
    """
    return {
        "max_conversation_messages": self.window_messages or UNBOUNDED,
        "compaction_trigger_chars": self.trigger_chars or UNBOUNDED,
        "compaction_keep_recent": self.keep_recent or _DEFAULT_KEEP_RECENT,
        "compaction_enabled": self.summarize,
    }

HistoryStrategy

Bases: Protocol

A named, swappable policy for keeping a conversation in the window.

directive 

directive() -> HistoryDirective

Return the inert decision. Must not touch configuration.

Source code in src/symfonic/services/conversation/history.py 
80
81
def directive(self) -> HistoryDirective:
    """Return the inert decision. Must not touch configuration."""

NullHistoryStrategy dataclass 

NullHistoryStrategy()

No management: history grows to the model's own limit.

SlidingWindowHistoryStrategy dataclass 

SlidingWindowHistoryStrategy(window_size: int)

Keep the most recent window_size messages; drop older ones.

SummarizingHistoryStrategy dataclass 

SummarizingHistoryStrategy(trigger_chars: int = _DEFAULT_TRIGGER_CHARS, keep_recent: int = _DEFAULT_KEEP_RECENT)

Summarize overflow into a running summary, keeping recent turns raw.

resolve_history 

resolve_history(strategy: HistoryStrategy | None, *, default: HistoryStrategy | None = None) -> HistoryDirective

Pick the governing directive: explicit strategy, then default, then the framework default.

A non-strategy argument is refused rather than duck-typed. Silently ignoring a misspelled strategy would leave the conversation unbounded while the caller believed it was windowed.

Source code in src/symfonic/services/conversation/history.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def resolve_history(
    strategy: HistoryStrategy | None, *, default: HistoryStrategy | None = None
) -> HistoryDirective:
    """Pick the governing directive: explicit strategy, then default, then the
    framework default.

    A non-strategy argument is refused rather than duck-typed. Silently
    ignoring a misspelled strategy would leave the conversation unbounded
    while the caller believed it was windowed.
    """
    for candidate in (strategy, default):
        if candidate is None:
            continue
        if not isinstance(candidate, HistoryStrategy):
            raise ConversationServiceError(
                f"{type(candidate).__name__} is not a history strategy"
            )
        return candidate.directive()
    return HistoryDirective.default()