Design and implement Schema-Guided Reasoning (SGR) pipelines. Translate domain expert mental checklists into structured reasoning schemas for LLMs.
\nSource: Rinat Abdullin — Schema-Guided Reasoning
\nSGR = guide LLM reasoning through predefined steps via constrained decoding. Instead of free-form text → enforce a schema that defines what steps, in which order, where to focus attention.
\nDomain expert mental checklist → Pydantic/Zod schema → Constrained decoding → Deterministic dispatch\nParse task from $ARGUMENTS:
Identify the reasoning cascade — interview the domain:
\nThis is the critical step. SGR quality = how well you translate the expert's mental checklist.
\nDesign the schema following SGR patterns:
\nclass NextStep(BaseModel):\n current_state: str # thinking space\n plan_remaining_steps: list[str] # 1-5 steps, only first used\n task_completed: bool # routing gate\n function: Union[Tool1, Tool2, ..., ReportCompletion] = Field(\n ..., description=\"execute first remaining step\"\n )\nclass Analysis(BaseModel):\n preliminary: str # initial assessment\n classification: Literal[\"a\", \"b\", \"c\"] # force categorization\n evidence: list[str] # cite sources\n gaps: list[GapItem] # structured findings\n verdict: Literal[\"pass\", \"partial\", \"fail\"] # final decision\n reasoning_for_verdict: str # explain after deciding\nclass SendEmail(BaseModel):\n tool: Literal[\"send_email\"] # discriminator\n recipient: str\n subject: str\n body: str\n\nclass SearchDB(BaseModel):\n tool: Literal[\"search_db\"]\n query: str\n\n# Union with Literal discriminator = deterministic routing\nAction = Union[SendEmail, SearchDB, ReportDone]\nApply SGR design rules (from references/sgr-rules.md):
Literal[\"pass\", \"fail\"] not strAnnotated[list[str], MinLen(1), MaxLen(5)]tool: Literal[\"name\"] for routingreasoning_for_X AFTER the enum field, not beforeAnnotated[int, Le(50)] — bake constraints into typesImplement the dispatch loop (if agent):
\nfor i in range(MAX_STEPS):\n response = client.beta.chat.completions.parse(\n model=MODEL,\n response_format=NextStep,\n messages=log,\n )\n job = response.choices[0].message.parsed\n\n if isinstance(job.function, ReportCompletion):\n break # done\n\n result = dispatch(job.function) # deterministic routing\n log.append(assistant_message(job))\n log.append(tool_result(result))\nAdd to project:
\nschemas/ or models/ directorydispatch.py or equivalentAudit mode (if $ARGUMENTS = \"audit\"):
## SGR Pipeline: {domain}\n\n**Pattern:** {NextStep | Analysis Cascade | Tool Dispatch}\n**Schemas:** {N} models\n**Tools:** {N} (if agent loop)\n\n### Reasoning Cascade\n{step 1} → {step 2} → ... → {decision/action}\n\n### Files\n- schemas/{name}.py — {N} models\n- dispatch.py — tool routing\n- tests/test_{name}.py — validation tests\nreferences/sgr-rules.md — design rules and anti-patternsreferences/sgr-demo.py — complete working example (Abdullin's CRM demo, 304 lines Python)references/sgr-patterns.md — cascade patterns for 6 domainsreferences/sgr-full-guide.md — full SGR guide with theory, code, tool calling internalsIn Rust, SGR is even stronger: #[serde(tag = \"tool\")] gives discriminated union dispatch at zero runtime cost. Enum variants = tools, serde deserialization = constrained decoding.
references/sgr-demo.py — minimal standalone example (304 lines, CRM agent)Cause: Tried to put everything in one model.\nFix: Split into analysis model + action model. Cascade, don't flatten.
\nCause: Model not supporting constrained decoding, or wrong API.\nFix: Use response_format=Schema (OpenAI), tools with schema (Anthropic). Check references/sgr-rules.md for provider-specific notes.
Cause: No task_completed gate or ReportCompletion tool.\nFix: Always include a completion signal in the Union. Cap loop iterations.