Technical Explanation of examples-agent-patterns-deterministic.py

This document provides a detailed explanation of the Python script examples-agent-patterns-deterministic.py.

Overview

The script demonstrates a deterministic, multi-agent workflow where a series of agents perform sequential tasks to generate a story. The flow is "deterministic" because the steps are predefined and executed in a specific order, with gates to control the progression based on intermediate results. This pattern is useful for creating reliable and predictable AI-powered workflows.

The workflow consists of three main stages: 1. Outline Generation: An agent generates a story outline based on user input. 2. Outline Verification: Another agent checks if the outline is of good quality and fits the desired genre (in this case, sci-fi). 3. Story Generation: A final agent writes the full story based on the verified outline.

The Agents

The script defines three distinct agents, each with a specific role:

1. story_outline_agent

This agent is responsible for creating a short story outline.

story_outline_agent = Agent(
    name="story_outline_agent",
    instructions="Generate a very short story outline based on the user's input.",
)

2. outline_checker_agent

This agent evaluates the generated outline. It checks for two conditions: - If the outline is of good quality. - If the story is a sci-fi story.

To ensure a structured output, it uses a Pydantic BaseModel called OutlineCheckerOutput.

class OutlineCheckerOutput(BaseModel):
    good_quality: bool
    is_scifi: bool


outline_checker_agent = Agent(
    name="outline_checker_agent",
    instructions="Read the given story outline, and judge the quality. Also, determine if it is a scifi story.",
    output_type=OutlineCheckerOutput,
)

3. story_agent

This agent takes the verified outline and writes a complete short story.

story_agent = Agent(
    name="story_agent",
    instructions="Write a short story based on the given outline.",
    output_type=str,
)

Deterministic Workflow

The main async function orchestrates the workflow:

  1. User Input: The script prompts the user for a story idea, with a fallback to a default prompt.
  2. Trace: The entire workflow is wrapped in a single trace for observability.
  3. Generate Outline: The story_outline_agent is run to generate the initial outline.
  4. Check Outline: The outline_checker_agent is run with the generated outline.
  5. Gate: The script checks the output from the outline_checker_agent.
  6. Write Story: If the outline passes both checks, the story_agent is run to write the final story, which is then printed to the console.
async def main():
    input_prompt = input_with_fallback(
        "What kind of story do you want? ",
        "Write a short sci-fi story.",
    )

    # Ensure the entire workflow is a single trace
    with trace("Deterministic story flow"):
        # 1. Generate an outline
        outline_result = await Runner.run(
            story_outline_agent,
            input_prompt,
        )
        print("Outline generated")

        # 2. Check the outline
        outline_checker_result = await Runner.run(
            outline_checker_agent,
            outline_result.final_output,
        )

        # 3. Add a gate to stop if the outline is not good quality or not a scifi story
        assert isinstance(outline_checker_result.final_output, OutlineCheckerOutput)
        if not outline_checker_result.final_output.good_quality:
            print("Outline is not good quality, so we stop here.")
            exit(0)

        if not outline_checker_result.final_output.is_scifi:
            print("Outline is not a scifi story, so we stop here.")
            exit(0)

        print("Outline is good quality and a scifi story, so we continue to write the story.")

        # 4. Write the story
        story_result = await Runner.run(
            story_agent,
            outline_result.final_output,
        )
        print(f"Story: {story_result.final_output}")

Script Output Analysis

The provided output file examples-agent-patterns-deterministic.py.txt shows the results of three separate executions of the script:

Execution 1

What kind of story do you want? Outline generated
Outline is not good quality, so we stop here.

In this run, the outline_checker_agent determined that the generated outline was not of good quality, so the script terminated before writing the story.

Execution 2

What kind of story do you want? Outline generated
Outline is good quality and a scifi story, so we continue to write the story.
Story: **The Boy Who Made a Spacecraft**

Ethan was not like most 12-year-olds in Maplewood...

Here, the outline passed both the quality and genre checks. The workflow proceeded, and the story_agent generated the story titled "The Boy Who Made a Spacecraft."

Execution 3

What kind of story do you want? Outline generated
Outline is not a scifi story, so we stop here.

In this final example, the outline was deemed to be of good quality, but it was not a sci-fi story. Therefore, the script stopped at the gate and did not generate the full story.

Full Script for Reference

from dotenv import load_dotenv
load_dotenv()

import asyncio

from pydantic import BaseModel

from agents import Agent, Runner, trace
from auto_mode import input_with_fallback

"""
This example demonstrates a deterministic flow, where each step is performed by an agent.
1. The first agent generates a story outline
2. We feed the outline into the second agent
3. The second agent checks if the outline is good quality and if it is a scifi story
4. If the outline is not good quality or not a scifi story, we stop here
5. If the outline is good quality and a scifi story, we feed the outline into the third agent
6. The third agent writes the story
"""

story_outline_agent = Agent(
    name="story_outline_agent",
    instructions="Generate a very short story outline based on the user's input.",
)


class OutlineCheckerOutput(BaseModel):
    good_quality: bool
    is_scifi: bool


outline_checker_agent = Agent(
    name="outline_checker_agent",
    instructions="Read the given story outline, and judge the quality. Also, determine if it is a scifi story.",
    output_type=OutlineCheckerOutput,
)

story_agent = Agent(
    name="story_agent",
    instructions="Write a short story based on the given outline.",
    output_type=str,
)


async def main():
    input_prompt = input_with_fallback(
        "What kind of story do you want? ",
        "Write a short sci-fi story.",
    )

    # Ensure the entire workflow is a single trace
    with trace("Deterministic story flow"):
        # 1. Generate an outline
        outline_result = await Runner.run(
            story_outline_agent,
            input_prompt,
        )
        print("Outline generated")

        # 2. Check the outline
        outline_checker_result = await Runner.run(
            outline_checker_agent,
            outline_result.final_output,
        )

        # 3. Add a gate to stop if the outline is not good quality or not a scifi story
        assert isinstance(outline_checker_result.final_output, OutlineCheckerOutput)
        if not outline_checker_result.final_output.good_quality:
            print("Outline is not good quality, so we stop here.")
            exit(0)

        if not outline_checker_result.final_output.is_scifi:
            print("Outline is not a scifi story, so we stop here.")
            exit(0)

        print("Outline is good quality and a scifi story, so we continue to write the story.")

        # 4. Write the story
        story_result = await Runner.run(
            story_agent,
            outline_result.final_output,
        )
        print(f"Story: {story_result.final_output}")


if __name__ == "__main__":
    asyncio.run(main())