Skip to content

symfonic.capabilities.memory.reflection

reflection

Post-consolidation insight extraction for the native memory capability.

This is deliberately separate from governance metacognition. Metacognition judges a draft before egress; reflection reads the completed exchange after ordinary memory publication and may persist a durable semantic insight. The two services therefore have different inputs, failure policy, and stage.

MemoryReflectionService 

MemoryReflectionService(model: ReflectionModelPort, *, prompt_template: str | None = None, scrubber: CredentialScrubber | None = None)

Render a reflection prompt and turn its JSON array into semantic records.

The service contains no store and never publishes by itself. The memory capability owns those effects at the FINALIZE rung, after ordinary extraction and its flush have completed.

Source code in src/symfonic/capabilities/memory/reflection.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def __init__(
    self,
    model: ReflectionModelPort,
    *,
    prompt_template: str | None = None,
    scrubber: CredentialScrubber | None = None,
) -> None:
    complete = getattr(model, "complete", None)
    if not callable(complete) or not inspect.iscoroutinefunction(complete):
        raise ConfigurationError(
            "MemoryReflectionService needs an async complete(prompt) model collaborator."
        )
    template = prompt_template or _DEFAULT_PROMPT
    if "{conversation}" not in template:
        raise ConfigurationError(
            "reflection prompt_template must contain {conversation}; otherwise the "
            "configured collaborator would receive no completed exchange."
        )
    self._model = model
    self._prompt_template = template
    self._scrubber = scrubber if scrubber is not None else CredentialScrubber()

prompt 

prompt(request: ReflectionRequest) -> tuple[str, tuple[str, ...]]

Return prompt bytes for the collaborator, with credentials redacted.

Source code in src/symfonic/capabilities/memory/reflection.py 
118
119
120
121
122
123
124
125
def prompt(self, request: ReflectionRequest) -> tuple[str, tuple[str, ...]]:
    """Return prompt bytes for the collaborator, with credentials redacted."""
    user = self._scrubber.scrub_text(request.user_message)
    assistant = self._scrubber.scrub_text(request.assistant_message)
    conversation = f"User: {user.text}\nAssistant: {assistant.text}"
    return self._prompt_template.format(
        conversation=conversation
    ), user.redactions + assistant.redactions

ReflectionModelPort

Bases: Protocol

The one collaborator a reflection pass needs.

complete async 

complete(prompt: str) -> Any

Return the model reply containing a JSON insight array.

Source code in src/symfonic/capabilities/memory/reflection.py 
54
55
async def complete(self, prompt: str) -> Any:
    """Return the model reply containing a JSON insight array."""

ReflectionPort

Bases: Protocol

The native reflection collaborator accepted by memory_capabilities.

reflect async 

reflect(request: ReflectionRequest) -> ReflectionResult

Extract durable insights from a completed exchange.

Source code in src/symfonic/capabilities/memory/reflection.py 
82
83
async def reflect(self, request: ReflectionRequest) -> ReflectionResult:
    """Extract durable insights from a completed exchange."""

ReflectionRequest dataclass 

ReflectionRequest(scope: MemoryScope, user_message: str, assistant_message: str, turn: int = 0)

The completed exchange a reflection pass may inspect.

ReflectionResult dataclass 

ReflectionResult(records: tuple[MemoryRecord, ...] = (), dropped: tuple[tuple[str, str], ...] = (), degraded: bool = False, reason: str = '')

Persistable reflection output, without leaking model payloads to traces.

validate_reflector 

validate_reflector(reflector: Any) -> None

Refuse a reflection collaborator that would be inert at runtime.

Source code in src/symfonic/capabilities/memory/reflection.py 
203
204
205
206
207
208
209
210
211
212
def validate_reflector(reflector: Any) -> None:
    """Refuse a reflection collaborator that would be inert at runtime."""
    if reflector is None:
        return
    reflect = getattr(reflector, "reflect", None)
    if not callable(reflect) or not inspect.iscoroutinefunction(reflect):
        raise ConfigurationError(
            "reflection must provide async reflect(ReflectionRequest) -> ReflectionResult; "
            "normally pass MemoryReflectionService(model, prompt_template=...)."
        )