Notebooks · Reliability and control · 54

Structured Output with Auto-Retry (v1.0.8)

Name: SHIPIT Agent
Author: SHIPIT

Get reliably-typed JSON / Pydantic output from any LLM, with three guarantees:

Open in Colab View on GitHub 8 code cells · 0 with saved output

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

In [ ]

python

from pathlib import Path
import sys

ROOT = Path.cwd().resolve().parent if Path.cwd().name == 'notebooks' else Path.cwd().resolve()
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from pydantic import BaseModel

from shipit_agent import Agent, StructuredOutput
from shipit_agent.parsers import parse_partial_json

1. Define your schema

Use a Pydantic model for typed output, or a JSON Schema dict for runtime-only validation. Both are accepted everywhere output_schema= is.

In [ ]

python

class Movie(BaseModel):
    title: str
    rating: float
    genre: str | None = None

Movie.model_json_schema()

2. Standalone extraction with `StructuredOutput`

We'll use a tiny scripted LLM here so the notebook runs without API keys. Swap in your real LLM (Bedrock, Anthropic, OpenAI, ...) and the code is identical.

In [ ]

python

from typing import Any, Iterator

class ScriptedLLM:
    def __init__(self, responses: list[str]):
        self.responses = responses
        self._i = 0
    def complete(self, *, messages: list[dict[str, Any]], **_):
        text = self.responses[self._i]
        self._i += 1
        return text
    def stream(self, *, messages: list[dict[str, Any]], **_) -> Iterator[str]:
        text = self.responses[self._i]
        self._i += 1
        for j in range(0, len(text), 4):
            yield text[j:j+4]

In [ ]

python

# First attempt succeeds — single LLM call, typed result.
llm = ScriptedLLM(['{"title": "Inception", "rating": 9.0}'])
so = StructuredOutput(llm=llm, schema=Movie, max_retries=0)
result = so.run('recommend a good thriller')
result.value

3. Validation-retry in action

When the model returns something that doesn't parse, StructuredOutput (and Agent.run with output_schema=) sends the error back to the same conversation and asks for a corrected response. Up to max_retries attempts.

In [ ]

python

# First reply is prose, not JSON. The retry path detects the parse error
# and asks the model to format properly. Note attempts == 2.
llm = ScriptedLLM([
    'I really like Inception, it has a 9 out of 10.',  # bad — pure prose
    '{"title": "Inception", "rating": 9.0}',           # corrected
])
so = StructuredOutput(llm=llm, schema=Movie, max_retries=2)
result = so.run('recommend a movie')
print(f'attempts: {result.attempts}')
print(f'parsed:   {result.value}')
print(f'first error: {result.history[0]["error"][:80]}...')

4. Streaming partial parse

Render structured output as tokens arrive — useful for live UIs that want to fill in fields one at a time.

In [ ]

python

llm = ScriptedLLM(['{"title": "Inception", "rating": 9.0, "genre": "thriller"}'])
so = StructuredOutput(llm=llm, schema=Movie, max_retries=0)
for partial in so.stream('recommend a movie'):
    print(repr(partial))

Notice each yield is a richer object than the previous: empty → just title → title + rating → fully-typed Movie. Frontends can render placeholders that fill in as the stream progresses.

5. Agent-loop integration: `Agent.run(output_schema=)`

The same retry path is wired into the main Agent and DeepAgent loops. Pass output_schema= and you get a typed result.parsed field; on parse failure the agent retries inside the same conversation.

In [ ]

python

from shipit_agent.llms.base import LLMResponse

class SimpleLLM:
    def __init__(self, replies: list[str]):
        self.replies = replies
        self._i = 0
    def complete(self, *, messages, **_):
        text = self.replies[min(self._i, len(self.replies) - 1)]
        self._i += 1
        return LLMResponse(content=text)

agent = Agent(llm=SimpleLLM([
    'I think Inception is great',
    '{"title": "Inception", "rating": 9.0}',
]))
result = agent.run(
    'recommend a movie',
    output_schema=Movie,
    max_validation_retries=2,
)
print('parsed:', result.parsed)
print('output text:', result.output)

6. Streaming partial JSON as a low-level utility

If you need to parse partial JSON outside the agent context (e.g. UI rendering of an arbitrary stream), parse_partial_json is exposed directly.

In [ ]

python

samples = [
    '{"name": "Al',
    '{"items": [1, 2, ',
    '{"x": {"y": 1, "z": ',
    'Sure! Here you go: {"a": 1}',
]
for s in samples:
    print(f'{s!r:<40} -> {parse_partial_json(s)!r}')

What this beats in LangChain

OutputFixingParser — LangChain's equivalent runs a separate LLM call with its own prompt. shipit's retry stays in the same conversation, so the model has full context of what it tried before. Cheaper and more accurate.
No streaming partial parser — LangChain's JsonOutputParser only emits parsed values for fully-valid JSON; the streaming partial here renders mid-stream.
Universal LLM contract — works with any object that exposes .complete(messages=...). No provider-specific structured output adapters needed.

Structured Output with Auto-Retry (v1.0.8)

Get reliably-typed JSON / Pydantic output from any LLM, with three guarantees:

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

In [ ]

python

from pathlib import Path
import sys

ROOT = Path.cwd().resolve().parent if Path.cwd().name == 'notebooks' else Path.cwd().resolve()
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from pydantic import BaseModel

from shipit_agent import Agent, StructuredOutput
from shipit_agent.parsers import parse_partial_json

1. Define your schema

Use a Pydantic model for typed output, or a JSON Schema dict for runtime-only validation. Both are accepted everywhere output_schema= is.

In [ ]

python

class Movie(BaseModel):
    title: str
    rating: float
    genre: str | None = None

Movie.model_json_schema()

2. Standalone extraction with `StructuredOutput`

We'll use a tiny scripted LLM here so the notebook runs without API keys. Swap in your real LLM (Bedrock, Anthropic, OpenAI, ...) and the code is identical.

In [ ]

python

from typing import Any, Iterator

class ScriptedLLM:
    def __init__(self, responses: list[str]):
        self.responses = responses
        self._i = 0
    def complete(self, *, messages: list[dict[str, Any]], **_):
        text = self.responses[self._i]
        self._i += 1
        return text
    def stream(self, *, messages: list[dict[str, Any]], **_) -> Iterator[str]:
        text = self.responses[self._i]
        self._i += 1
        for j in range(0, len(text), 4):
            yield text[j:j+4]

In [ ]

python

# First attempt succeeds — single LLM call, typed result.
llm = ScriptedLLM(['{"title": "Inception", "rating": 9.0}'])
so = StructuredOutput(llm=llm, schema=Movie, max_retries=0)
result = so.run('recommend a good thriller')
result.value

3. Validation-retry in action

In [ ]

python

# First reply is prose, not JSON. The retry path detects the parse error
# and asks the model to format properly. Note attempts == 2.
llm = ScriptedLLM([
    'I really like Inception, it has a 9 out of 10.',  # bad — pure prose
    '{"title": "Inception", "rating": 9.0}',           # corrected
])
so = StructuredOutput(llm=llm, schema=Movie, max_retries=2)
result = so.run('recommend a movie')
print(f'attempts: {result.attempts}')
print(f'parsed:   {result.value}')
print(f'first error: {result.history[0]["error"][:80]}...')

4. Streaming partial parse

Render structured output as tokens arrive — useful for live UIs that want to fill in fields one at a time.

In [ ]

python

llm = ScriptedLLM(['{"title": "Inception", "rating": 9.0, "genre": "thriller"}'])
so = StructuredOutput(llm=llm, schema=Movie, max_retries=0)
for partial in so.stream('recommend a movie'):
    print(repr(partial))

Notice each yield is a richer object than the previous: empty → just title → title + rating → fully-typed Movie. Frontends can render placeholders that fill in as the stream progresses.

5. Agent-loop integration: `Agent.run(output_schema=)`

In [ ]

python

from shipit_agent.llms.base import LLMResponse

class SimpleLLM:
    def __init__(self, replies: list[str]):
        self.replies = replies
        self._i = 0
    def complete(self, *, messages, **_):
        text = self.replies[min(self._i, len(self.replies) - 1)]
        self._i += 1
        return LLMResponse(content=text)

agent = Agent(llm=SimpleLLM([
    'I think Inception is great',
    '{"title": "Inception", "rating": 9.0}',
]))
result = agent.run(
    'recommend a movie',
    output_schema=Movie,
    max_validation_retries=2,
)
print('parsed:', result.parsed)
print('output text:', result.output)

6. Streaming partial JSON as a low-level utility

If you need to parse partial JSON outside the agent context (e.g. UI rendering of an arbitrary stream), parse_partial_json is exposed directly.

In [ ]

python

samples = [
    '{"name": "Al',
    '{"items": [1, 2, ',
    '{"x": {"y": 1, "z": ',
    'Sure! Here you go: {"a": 1}',
]
for s in samples:
    print(f'{s!r:<40} -> {parse_partial_json(s)!r}')

What this beats in LangChain

OutputFixingParser — LangChain's equivalent runs a separate LLM call with its own prompt. shipit's retry stays in the same conversation, so the model has full context of what it tried before. Cheaper and more accurate.
No streaming partial parser — LangChain's JsonOutputParser only emits parsed values for fully-valid JSON; the streaming partial here renders mid-stream.
Universal LLM contract — works with any object that exposes .complete(messages=...). No provider-specific structured output adapters needed.

Structured Output with Auto-Retry (v1.0.8)

1. Define your schema

2. Standalone extraction with StructuredOutput

3. Validation-retry in action

4. Streaming partial parse

5. Agent-loop integration: Agent.run(output_schema=)

6. Streaming partial JSON as a low-level utility

What this beats in LangChain

Structured Output with Auto-Retry (v1.0.8)

1. Define your schema

2. Standalone extraction with StructuredOutput

3. Validation-retry in action

4. Streaming partial parse

5. Agent-loop integration: Agent.run(output_schema=)

6. Streaming partial JSON as a low-level utility

What this beats in LangChain

2. Standalone extraction with `StructuredOutput`

5. Agent-loop integration: `Agent.run(output_schema=)`

2. Standalone extraction with `StructuredOutput`

5. Agent-loop integration: `Agent.run(output_schema=)`