AI Agent Beginner Tutorial: 10 - Workflows and Development Frameworks

Place four patterns in the brief flow

Sequence: validate task → retrieve → generate → validate. Each step depends on the prior output.

Routing: request a topic when missing, report an evidence gap when retrieval is empty, and generate only with sufficient evidence.

Parallelism: local and permitted remote retrieval can run together, with branches reading shared input only.

Dynamic planning: the model selects release notes, API documentation, or security advisories to resolve evidence conflicts, under step and tool budgets.

Merge parallel branches with a reducer

from dataclasses import dataclass

@dataclass(frozen=True)
class Source:
    source_id: str
    score: float
    text: str

def merge_sources(groups: list[list[Source]], limit: int = 6) -> list[Source]:
    best: dict[str, Source] = {}
    for group in groups:
        for source in group:
            current = best.get(source.source_id)
            if current is None or source.score > current.score:
                best[source.source_id] = source
    return sorted(best.values(), key=lambda item: (-item.score, item.source_id))[:limit]

Branches do not mutate one shared state dictionary. They return isolated lists, and a pure function deduplicates and sorts deterministically. Changing completion order cannot change the final source set.

Route on structured fields

def choose_next(task_valid: bool, evidence_count: int,
                citations_valid: bool) -> str:
    if not task_valid:
        return "request_clarification"
    if evidence_count == 0:
        return "report_evidence_gap"
    if not citations_valid:
        return "repair_citations"
    return "request_save_approval"

Free text fits presentation, while enums fit routing. Parsing a model sentence such as “the sources may be enough” to permit a write makes tests and recovery fragile.

Retry from safe nodes

Pure read nodes can rerun. Writes with idempotency keys query status first. Irreversible actions wait for human handling. Retries inherit global turn, tool-call, and cost budgets rather than resetting counters per node.

Fault injection covers four positions: one parallel search times out, one branch returns bad data, citations fail after merging, and the process exits before saving. Recovery executes only unconfirmed safe nodes.

Map the project's current control flow

validate_task
      │
      ▼
local_search ─┐
              ├─▶ merge_sources ─▶ assess_evidence
remote_search ┘                         │
                           ┌─────────────┴─────────────┐
                           ▼                           ▼
                     search_more                 draft_and_check
                                                        │
                                                        ▼
                                                 wait_for_approval

This graph is more complex than the minimal loop, and the hand-written runtime now carries generic graph state, sessions, and tracing. With the workflow organized, the following section migrates to the OpenAI Agents SDK under behavioral-test protection and places LangGraph in the explicit-graph and checkpointing design space.

Begin migration with a responsibility map

Using a framework before writing a loop hides the control flow behind its APIs. The preceding implementation already covers model loops, tool dispatch, errors, state, and approvals, so each generic responsibility taken over by the SDK is now visible.

Manual runtime                       Agents SDK
──────────────                       ──────────
for turn in range(...)       ───▶   Runner.run
parse function_call           ───▶   function_tool
assemble history              ───▶   Session / response chain
pause object                  ───▶   interruption + RunState
event log                     ───▶   Trace

Task spec, permissions, business budgets, acceptance ───▶ stay in the app

Freeze a Runtime protocol first

The domain layer defines run(task) -> RunResult. Manual and SDK versions both return terminal status, final output, source IDs, tool events, and cost information. Tests do not depend on SDK-internal class names.

Do not change prompts, tool semantics, or task budgets during migration. Otherwise, result differences cannot be attributed to the runtime.

Wrap existing tools with function_tool

import os
from agents import Agent, Runner, SQLiteSession, function_tool

SOURCES = {"agent-loop#1": "A tool result returns to the next model turn."}

@function_tool
def get_document(source_id: str) -> dict:
    """Read one approved source by its exact source ID."""
    text = SOURCES.get(source_id)
    if text is None:
        return {"ok": False, "error": "source_not_found"}
    return {"ok": True, "source_id": source_id, "text": text}

brief_agent = Agent(
    name="Research brief agent",
    instructions=(
        "Use approved sources. Preserve source IDs. "
        "Return insufficient_evidence when support is missing."
    ),
    model=os.environ["OPENAI_MODEL"],
    tools=[get_document],
)

Tools continue returning application-defined domain results. The domain layer receives no SDK-specific object, so a future runtime change does not require rewriting retrieval and acceptance.

Combine Runner and Session

async def run_brief(thread_id: str, question: str) -> str:
    session = SQLiteSession(thread_id, "brief_sessions.db")
    result = await Runner.run(
        brief_agent,
        question,
        session=session,
        max_turns=6,
    )
    return result.final_output

Sessions, full history, and previousResponseId can all continue context; choose one primary strategy. Sending full history while also using a session duplicates messages. Bind thread_id to tenant and user scope instead of accepting a short front-end string as a global key.

Protect migration with shared behavioral tests

Run the same cases against ManualRuntime and AgentsSdkRuntime: search then read, unknown source, insufficient evidence, turn exhaustion, pending approval, and save rejection. Compare domain results and environment side effects rather than exact internal trace event names.

                 ┌──────────────────┐
Task fixture ───▶ │ Runtime contract │
                 └───────┬──────────┘
                         │
             ┌───────────┴───────────┐
             ▼                       ▼
      ManualRuntime           AgentsSdkRuntime
             │                       │
             └───────────┬───────────┘
                         ▼
                 Same outcome checker

Keep LangGraph as an extension

The project's dynamic surface remains small, so the Agents SDK is sufficient. When a system needs explicit nodes, complex branches, graph-level checkpoints, or visible state, rewrite retrieve → assess → draft → approve in LangGraph while reusing BriefTask, ToolRegistry, and outcome checkers.

Framework selection records which code it replaces and which state formats and upgrade costs it introduces. A framework name does not improve agent quality by itself.

Set the migration acceptance line

• Manual and SDK versions pass the same behavioral tests.
• A session continues only within the same thread.
• Application budgets and Runner max_turns both apply.
• Approval, path permissions, and outcome acceptance remain outside prompts.
• SDK upgrades run regressions before tutorial code changes.

After runtime migration, Chapter 11 connects out-of-process capabilities. MCP lets a host discover tools, resources, and prompts exposed by a server, while the application retains its allowlists, approvals, and untrusted-content boundary.