Skip to content

symfonic.capabilities.critic.reflection

reflection

The hidden reflection call behind the critic verdict (CORE-403).

One prompt, one model round-trip, one parsed verdict -- ported from MetacognitiveMiddleware's reflection half. The prompt text and the approve-on-malformed rule are verbatim; the provider arrives injected and the rules arrive through :class:ProceduralRulesSource, so this module never touches the orchestrator.

CriticPreference dataclass 

CriticPreference(priority: float, statement: str, sensitive_tags: frozenset[str] = frozenset())

One rule the draft must respect, as values (no domain import).

ProceduralRulesSource

Bases: Protocol

Per-turn procedural rules, supplied by the composition root.

retrieve async 

retrieve(query: str, top_k: int) -> Sequence[str]

Return up to top_k rule contents for query.

Source code in src/symfonic/capabilities/critic/reflection.py 
74
75
76
async def retrieve(self, query: str, top_k: int) -> Sequence[str]:
    """Return up to ``top_k`` rule contents for ``query``."""
    ...

parse_reflection_output 

parse_reflection_output(raw: str) -> tuple[str, str | None, list[str]]

Ported verbatim: structured verdict, approve on malformed output.

Source code in src/symfonic/capabilities/critic/reflection.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def parse_reflection_output(raw: str) -> tuple[str, str | None, list[str]]:
    """Ported verbatim: structured verdict, approve on malformed output."""
    try:
        cleaned = raw.strip()
        if cleaned.startswith("```"):
            lines = cleaned.split("\n")
            lines = [ln for ln in lines if not ln.strip().startswith("```")]
            cleaned = "\n".join(lines).strip()

        data = json.loads(cleaned)
        verdict_raw = data.get("verdict", "approve")
        if verdict_raw not in ("approve", "revise", "refuse"):
            return "approve", None, []

        text = None
        if verdict_raw == "revise":
            text = data.get("revised_draft")
        elif verdict_raw == "refuse":
            text = data.get("refusal_message")

        matched = data.get("matched_preferences", [])
        if not isinstance(matched, list):
            matched = []

        return verdict_raw, text, matched
    except (json.JSONDecodeError, AttributeError, TypeError):
        return "approve", None, []

reflect_on_draft async 

reflect_on_draft(provider: Any, model_config: Any, template: str, draft: str, preferences: tuple[CriticPreference, ...], rules: list[str], hit: str | None) -> tuple[str, str | None, list[str]]

One hidden reflection call; approve on any failure (legacy rule).

Source code in src/symfonic/capabilities/critic/reflection.py 
 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
105
106
107
async def reflect_on_draft(
    provider: Any,
    model_config: Any,
    template: str,
    draft: str,
    preferences: tuple[CriticPreference, ...],
    rules: list[str],
    hit: str | None,
) -> tuple[str, str | None, list[str]]:
    """One hidden reflection call; approve on any failure (legacy rule)."""
    try:
        prefs_text = "\n".join(
            f"- [{p.priority}] {p.statement}"
            for p in sorted(preferences, key=lambda p: p.priority, reverse=True)
        ) or "(none)"
        rules_text = "\n".join(f"- {rule}" for rule in rules) or "(none)"
        prompt = template.format(
            draft=draft,
            core_preferences=prefs_text,
            procedural_rules=rules_text,
            sensitive_tag_hit=hit or "none",
        )
        model = provider.get_chat_model(model_config)
        result = await model.ainvoke([HumanMessage(content=prompt)])
        raw = result.content if hasattr(result, "content") else str(result)
        return parse_reflection_output(str(raw))
    except Exception:
        logger.warning("Critic reflection failed, falling back to approve", exc_info=True)
        return "approve", None, []