Skip to content

Deterministic Pattern

Source code in OpAgentsOlympus/practice/deterministic_pattern.py
OpAgentsOlympus/practice/deterministic_pattern.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import asyncio

from pydantic import BaseModel
from config import config
from agents import Agent, Runner

"""
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("What kind of story do you want? ")

    # 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, run_config=config
    )
    print("Outline generated")

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

    # 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, run_config=config
    )
    print(f"Story: {story_result.final_output}")


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