Writing
Pipelines
Workflows
Structured output
You already know the audience, the goal, and the points to hit — what costs the afternoon is the first draft, and the one-shot draft always misses two of the points.
Paste into Claude, Codex, Cursor, or another coding agent to scaffold this harness.
# Context: Vidbyte SDK
Vidbyte SDK is a Python package (import as `vidbyte`) for building agent workflows yourself.
Use it when you want to own the agent loop: system prompts, model/provider choice, tools, managed context, middleware, pipelines, tracing, evals, and multi-agent composition.
Core mental model:
1. Create an Agent or BaseAgent.
2. Attach a system prompt, provider/model, optional tools, context, middleware, and runtime choices.
3. Call run() or arun().
4. Let the SDK handle message assembly, tool schemas/calls, iteration limits, and optional pipelines.
Public package boundary: reusable local agent infrastructure. It does not ship private Vidbyte learning models or proprietary platform internals.
Every step adds to the same file. By the end you have one runnable writer.py.
The checklist field matters as much as the markdown: it is where the model puts claims and links it could not verify, instead of quietly asserting them.
+ a typed draft with an editor checklist
from pydantic import BaseModel, Field
class Draft(BaseModel):
title: str = Field(description="The headline for the piece.")
markdown: str = Field(description="The full draft, ready for a human to edit.")
editor_checklist: list[str] = Field(description="Claims or links a human must verify.")A structured brief is what makes two weeks of drafts comparable. Rendering it through one function means the writer sees the same fields every time, plus any notes from a previous attempt.
+ the same inputs every run
brief = {
"format": "blog post",
"audience": "developers evaluating agent frameworks",
"goal": "explain why a controllable agent loop matters",
"tone": "clear, practical, non-hype",
"length": "600-800 words",
"must_include": ["tools and verification", "how it differs from one-shot chat", "one concrete example"],
"must_avoid": ["fake customer logos", "unverified benchmarks"],
}
def format_brief(brief: dict, feedback: tuple[str, ...] = ()) -> str:
"""Render the brief, plus any reviewer notes from the previous attempt."""
lines = [f"{key}: {value}" for key, value in brief.items()]
if feedback:
lines.append("Fix these reviewer notes: " + "; ".join(feedback))
return "\n".join(lines)One agent with one job. Because it carries the Draft schema, its output is an object with a title and a body — not a blob you have to split apart.
+ the writer
from vidbyte import Agent
writer = Agent(
name="content-writer",
system_prompt=(
"Write marketing and product drafts from a brief. "
"Follow the audience, tone, format, and length exactly. "
"Never invent product claims the brief did not give you."
),
provider="openai",
model_name="gpt-4.1",
output_schema=Draft,
)A separate agent that only compares the draft to the brief catches misses without rewriting everything. The approved boolean is what the loop in step 5 branches on.
+ the reviewer, and a machine-readable verdict
class Review(BaseModel):
missing_points: list[str] = Field(description="Must-include points the draft skipped.")
tone_issues: list[str] = Field(description="Places the tone drifts from the brief.")
invented_claims: list[str] = Field(description="Claims not supported by the brief.")
approved: bool = Field(description="True only when nothing above needs fixing.")
reviewer = Agent(
name="brief-reviewer",
system_prompt=(
"Compare the draft to the brief. "
"List missing points, tone mismatches, and invented claims. "
"Approve only when all three lists are empty."
),
provider="openai",
model_name="gpt-4.1",
output_schema=Review,
)A pipeline threads one agent's text output into the next agent's prompt. It is the right shape when you just want a review printed alongside the draft — and if that is all you need, stop here.
+ the simple write-then-review chain
from vidbyte import SequentialPipeline
review_chain = SequentialPipeline([writer, reviewer])
printed_review = await review_chain.run(format_brief(brief))A pipeline runs forward once and hands back text. When the review has to feed back into a rewrite, you need a graph: a stage that drafts, a validator that reads approved, and an edge that routes a rejection back to the stage.
+ a bounded rewrite loop
from dataclasses import dataclass, field, replace
from vidbyte import (
CallableStage,
CallableValidator,
MachineStatus,
StageResult,
StateGraph,
StateMachineSettings,
ValidationResult,
)
@dataclass(frozen=True)
class WriterState:
brief: dict
draft: Draft | None = None
feedback: tuple[str, ...] = field(default_factory=tuple)
async def draft_stage(ctx):
# Writes a draft, then asks the reviewer to check it against the brief.
prompt = format_brief(ctx.state.brief, feedback=ctx.state.feedback)
written = (await writer.arun(prompt)).structured
verdict = (await reviewer.arun(f"Brief: {ctx.state.brief}\n\nDraft:\n{written.markdown}")).structured
notes = (*verdict.missing_points, *verdict.tone_issues, *verdict.invented_claims)
return StageResult(
replace(ctx.state, draft=written, feedback=notes),
outcome="success" if verdict.approved else "needs_revision",
)
def review_passed(ctx):
# Passes only when the reviewer left no notes on the candidate draft.
if not ctx.candidate_state.feedback:
return ValidationResult.passed()
return ValidationResult.rejected("needs_revision", "Address the reviewer notes.")
graph = StateGraph(WriterState, name="draft-until-approved")
graph.add_stage("draft", CallableStage(draft_stage), validators=(CallableValidator(review_passed),))
graph.add_terminal("approved", status=MachineStatus.SUCCEEDED)
graph.set_entry("draft")
graph.add_transition("draft", "approved")
graph.add_transition("draft", "draft", on="needs_revision")
machine = graph.compile(settings=StateMachineSettings(max_transitions=6))The machine returns the last committed state along with how it terminated. Hitting the transition ceiling is a real outcome you should check, not an exception to ignore.
+ the run, and the approved draft
result = await machine.arun(WriterState(brief=brief))
print("status:", result.status)
print(result.state.draft.title)
print(result.state.draft.markdown)
print("verify before publishing:", result.state.draft.editor_checklist)The seven steps above, in one file. The workflow is what runs; the pipeline from step 5 is kept as the one-pass alternative.
writer.py — complete
"""Content Writer Agent - drafts from a brief and revises until a reviewer approves."""
import asyncio
from dataclasses import dataclass, field, replace
from pydantic import BaseModel, Field
from vidbyte import (
Agent,
CallableStage,
CallableValidator,
MachineStatus,
StageResult,
StateGraph,
StateMachineSettings,
ValidationResult,
)
class Draft(BaseModel):
title: str = Field(description="The headline for the piece.")
markdown: str = Field(description="The full draft, ready for a human to edit.")
editor_checklist: list[str] = Field(description="Claims or links a human must verify.")
class Review(BaseModel):
missing_points: list[str] = Field(description="Must-include points the draft skipped.")
tone_issues: list[str] = Field(description="Places the tone drifts from the brief.")
invented_claims: list[str] = Field(description="Claims not supported by the brief.")
approved: bool = Field(description="True only when nothing above needs fixing.")
writer = Agent(
name="content-writer",
system_prompt=(
"Write marketing and product drafts from a brief. "
"Follow the audience, tone, format, and length exactly. "
"Never invent product claims the brief did not give you."
),
provider="openai",
model_name="gpt-4.1",
output_schema=Draft,
)
reviewer = Agent(
name="brief-reviewer",
system_prompt=(
"Compare the draft to the brief. "
"List missing points, tone mismatches, and invented claims. "
"Approve only when all three lists are empty."
),
provider="openai",
model_name="gpt-4.1",
output_schema=Review,
)
brief = {
"format": "blog post",
"audience": "developers evaluating agent frameworks",
"goal": "explain why a controllable agent loop matters",
"tone": "clear, practical, non-hype",
"length": "600-800 words",
"must_include": ["tools and verification", "how it differs from one-shot chat", "one concrete example"],
"must_avoid": ["fake customer logos", "unverified benchmarks"],
}
def format_brief(brief: dict, feedback: tuple[str, ...] = ()) -> str:
"""Render the brief, plus any reviewer notes from the previous attempt."""
lines = [f"{key}: {value}" for key, value in brief.items()]
if feedback:
lines.append("Fix these reviewer notes: " + "; ".join(feedback))
return "\n".join(lines)
@dataclass(frozen=True)
class WriterState:
brief: dict
draft: Draft | None = None
feedback: tuple[str, ...] = field(default_factory=tuple)
async def draft_stage(ctx):
# Writes a draft, then asks the reviewer to check it against the brief.
prompt = format_brief(ctx.state.brief, feedback=ctx.state.feedback)
written = (await writer.arun(prompt)).structured
verdict = (await reviewer.arun(f"Brief: {ctx.state.brief}\n\nDraft:\n{written.markdown}")).structured
notes = (*verdict.missing_points, *verdict.tone_issues, *verdict.invented_claims)
return StageResult(
replace(ctx.state, draft=written, feedback=notes),
outcome="success" if verdict.approved else "needs_revision",
)
def review_passed(ctx):
# Passes only when the reviewer left no notes on the candidate draft.
if not ctx.candidate_state.feedback:
return ValidationResult.passed()
return ValidationResult.rejected("needs_revision", "Address the reviewer notes.")
graph = StateGraph(WriterState, name="draft-until-approved")
graph.add_stage("draft", CallableStage(draft_stage), validators=(CallableValidator(review_passed),))
graph.add_terminal("approved", status=MachineStatus.SUCCEEDED)
graph.set_entry("draft")
graph.add_transition("draft", "approved")
graph.add_transition("draft", "draft", on="needs_revision")
machine = graph.compile(settings=StateMachineSettings(max_transitions=6))
async def main() -> None:
result = await machine.arun(WriterState(brief=brief))
print("status:", result.status)
print(result.state.draft.title)
print(result.state.draft.markdown)
print("verify before publishing:", result.state.draft.editor_checklist)
if __name__ == "__main__":
asyncio.run(main())