Research
Structured output
Tools
Sources
Answering one research question well costs an hour of open tabs, and at the end you still cannot say which claim came from where.
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 research_agent.py.
Write the answer's shape before writing any agent code. Declaring a Pydantic model here is what turns the output from prose into data you can render, store, or diff.
+ the shape of a finished brief
from pydantic import BaseModel, Field
class Finding(BaseModel):
claim: str = Field(description="One specific finding, stated plainly.")
source_url: str = Field(description="The URL a tool returned this claim from.")
class ResearchBrief(BaseModel):
summary: str = Field(description="Two or three sentences a busy reader can skim.")
findings: list[Finding] = Field(min_length=3, description="Evidence, one claim per item.")
open_questions: list[str] = Field(description="What the evidence did not settle.")A tool is an ordinary Python function with a decorator. The SDK reads the signature and docstring to build the schema it shows the model, so you never hand-write a JSON tool definition.
+ search_web and fetch_page
from vidbyte import tool
@tool
def search_web(query: str) -> list[dict]:
"""Search the web and return result titles, snippets, and URLs."""
# Call your search provider here and return one dict per result.
...
@tool
def fetch_page(url: str) -> str:
"""Fetch one URL and return its readable text."""
...Output contracts are deterministic floors the runtime checks when the agent tries to finish. An unmet floor sends the agent back to work, which holds better than asking politely in the system prompt.
+ a floor on effort before finishing
from vidbyte.agents import AgentLoopSettings
from vidbyte.agents.contracts import MinDistinctTools, MinToolCalls
loop_settings = AgentLoopSettings(
max_iterations=10,
output_contracts=(MinToolCalls(2), MinDistinctTools(2)),
max_contract_rejections=2,
)Trace.debug collects every model call and tool call into a plain list you can print. Use it while you are tuning the prompt, then swap in a hosted tracer later without touching the agent.
+ a readable record of the run
from vidbyte import Trace
events: list[dict] = []
trace = Trace.debug(events)Everything built so far is passed in one place. The compact_tool_outputs preset shortens tool results as they age, so a few fetched pages do not crowd the question out of the context window.
+ the agent, wired from the four pieces above
from vidbyte import Agent, ContextWindow
agent = Agent(
name="research-reporter",
system_prompt=(
"Research the question with the tools before answering. "
"Every finding must name the source it came from. "
"Say plainly when the evidence is thin."
),
provider="openai",
model_name="gpt-4.1",
tools=[search_web, fetch_page],
output_schema=ResearchBrief,
algorithm=ContextWindow.preset.compact_tool_outputs,
agent_loop_settings=loop_settings,
trace=trace,
)Because a schema was declared, reply.structured is a validated ResearchBrief — never None, never a string you have to parse. get_cost_usd totals what the run actually spent.
+ the run, and what it cost
reply = await agent.arun("What are the main approaches teams use for AI code review in 2026?")
brief = reply.structured
print(brief.summary)
for finding in brief.findings:
print(f"- {finding.claim} ({finding.source_url})")
print("open questions:", brief.open_questions)
print("run cost (usd):", agent.get_cost_usd())The six steps above, in one file.
research_agent.py — complete
"""Research Report Agent - gathers evidence with tools, returns a typed brief."""
import asyncio
from pydantic import BaseModel, Field
from vidbyte import Agent, ContextWindow, Trace, tool
from vidbyte.agents import AgentLoopSettings
from vidbyte.agents.contracts import MinDistinctTools, MinToolCalls
class Finding(BaseModel):
claim: str = Field(description="One specific finding, stated plainly.")
source_url: str = Field(description="The URL a tool returned this claim from.")
class ResearchBrief(BaseModel):
summary: str = Field(description="Two or three sentences a busy reader can skim.")
findings: list[Finding] = Field(min_length=3, description="Evidence, one claim per item.")
open_questions: list[str] = Field(description="What the evidence did not settle.")
@tool
def search_web(query: str) -> list[dict]:
"""Search the web and return result titles, snippets, and URLs."""
...
@tool
def fetch_page(url: str) -> str:
"""Fetch one URL and return its readable text."""
...
events: list[dict] = []
agent = Agent(
name="research-reporter",
system_prompt=(
"Research the question with the tools before answering. "
"Every finding must name the source it came from. "
"Say plainly when the evidence is thin."
),
provider="openai",
model_name="gpt-4.1",
tools=[search_web, fetch_page],
output_schema=ResearchBrief,
algorithm=ContextWindow.preset.compact_tool_outputs,
agent_loop_settings=AgentLoopSettings(
max_iterations=10,
output_contracts=(MinToolCalls(2), MinDistinctTools(2)),
max_contract_rejections=2,
),
trace=Trace.debug(events),
)
async def main() -> None:
reply = await agent.arun("What are the main approaches teams use for AI code review in 2026?")
brief = reply.structured
print(brief.summary)
for finding in brief.findings:
print(f"- {finding.claim} ({finding.source_url})")
print("open questions:", brief.open_questions)
print("run cost (usd):", agent.get_cost_usd())
if __name__ == "__main__":
asyncio.run(main())