ZerdaZerda

Cookbook

On this page

Practical recipes built from the actual behavior of output_parser.

Recipe: Fall back through multiple formats without the full orchestrator

If you don't need retry/safety/caching but do want "try format A, then format B", use ChainedParser instead of the full OutputParser:

Python · 6 lines
from output_parser.base_parser import ChainedParser
from output_parser.json_parser import JSONOutputParser
from output_parser.list_parser import ListOutputParser

parser = ChainedParser([JSONOutputParser(), ListOutputParser()])
result = parser.parse(llm_output)  # tries JSON first, falls back to list extraction

Recipe: Custom business-rule validation on top of a Pydantic schema

OutputValidator accepts arbitrary ValidationRules alongside Pydantic structural validation:

Python · 16 lines
from output_parser.validator import OutputValidator, ValidationRule
from output_parser.schema import ValidationStatus, AnswerSchema

validator = OutputValidator(
    pydantic_model=AnswerSchema,
    rules=[
        ValidationRule(
            "sources_present_if_confident",
            lambda d: not (d.get("confidence", 0) > 0.8 and not d.get("sources")),
            "High-confidence answers should cite at least one source",
            severity=ValidationStatus.WARNING,
        ),
    ],
)
results = validator.validate({"answer": "Paris", "confidence": 0.95, "sources": []})
failures = [r for r in results if r.status == ValidationStatus.FAILED]

Note: OutputParser itself has no constructor parameter for injecting custom ValidationRules — if you need this combined with the full pipeline, build your own OutputValidator and either call it manually after OutputParser.parse(), or construct OutputParser's sub-pieces yourself rather than using the class as-is.

Recipe: Safety-check raw LLM output before it ever reaches your app logic

Use OutputValidator's safety layer standalone, independent of parsing:

Python · 7 lines
from output_parser.validator import OutputValidator

validator = OutputValidator(enable_safety=True)
results = validator.validate(data=None, raw_text=llm_raw_response)
leak_or_injection = [r for r in results if r.status == ValidationStatus.FAILED]
if leak_or_injection:
    reject_response(llm_raw_response)

Recipe: Repair malformed JSON without re-prompting the LLM

Python · 11 lines
from output_parser.fixer import OutputFixer
from output_parser.schema import OutputFormat
import json

fixer = OutputFixer(required_fields=["answer"], field_defaults={"answer": ""})
fixed_text, strategy = fixer.fix(
    "{answer: 'Paris', confidence: 0.9,}",  # unquoted key, single quotes, trailing comma
    expected_format=OutputFormat.JSON,
)
data = json.loads(fixed_text)
print(strategy)  # FixStrategy.REGEX_REPAIR

Recipe: Full production pipeline — detect, parse, repair, retry, validate, log

Python · 26 lines
from output_parser import OutputParser, ParseMode
from output_parser.schema import AnswerSchema

def call_llm(prompt: str) -> str:
    ...  # your model call here

parser = OutputParser(
    schema=AnswerSchema,
    llm_fn=call_llm,
    max_retries=3,
    mode=ParseMode.LENIENT,
    original_prompt=user_question,
    enable_safety=True,
)

result = parser.parse(raw_llm_response)

if result.ok:
    answer: AnswerSchema = result.data
    respond_to_user(answer.answer, sources=answer.sources)
else:
    log_failure(result.trace.summary())
    respond_to_user("Sorry, I couldn't process that.")

# Always log the trace for observability, regardless of outcome
telemetry.record("output_parser.parse", result.trace.summary())

Recipe: Tool/function-call extraction from ReAct-style agent output

Python · 12 lines
from output_parser import create_tool_call_parser

parser = create_tool_call_parser(llm_fn=call_llm)
agent_output = '''
Thought: I need to look up the weather.
Action: get_weather
Action Input: {"city": "Paris"}
'''
result = parser.parse(agent_output)
print(result.data)
# {'tool_name': 'get_weather', 'arguments': {'city': 'Paris'},
#  'thought': 'I need to look up the weather.'}

Recipe: Async-friendly single-format parsing

Python · 10 lines
import asyncio
from output_parser.json_parser import JSONOutputParser

parser = JSONOutputParser()

async def handle_response(text: str):
    data = await parser.async_parse(text)
    return data

asyncio.run(handle_response('{"answer": "ok"}'))

Note: OutputParser (the orchestrator) does not define an async_parse method — only BaseOutputParser subclasses (the single-format toolkit) get it. For async use of the orchestrator, wrap the call yourself:

Python · 2 lines
import asyncio
result = await asyncio.get_running_loop().run_in_executor(None, parser.parse, raw_text)

Recipe: CSV table with typed columns from an LLM data-extraction prompt

Python · 11 lines
from output_parser.csv_parser import CSVOutputParser

parser = CSVOutputParser(
    required_columns=["product", "price", "in_stock"],
    column_types={"price": float, "in_stock": lambda v: v.strip().lower() == "true"},
)
text = '''
```csv
product,price,in_stock
Widget,9.99,true
Gadget,19.99,false

''' rows = parser.parse(text)

Related pages