Workflows - Orkestra

Workflows chain agents in sequence and support conditional logic via step handlers.

Create and chain

researcher = client.Agent("Researcher", "Finds info", "openai", "gpt-4-turbo", "OPENAI_API_KEY")
summarizer = client.Agent("Summarizer", "Summarizes", "openai", "gpt-3.5-turbo", "OPENAI_API_KEY")

wf = client.Workflow().add(researcher).add(summarizer)
result = wf.run("What are the key differences between nuclear fission and fusion?")

Structured output for final step

from pydantic import BaseModel
class KeyDifferences(BaseModel):
  title: str
  points: list[str]
  summary: str

final = wf.run("Explain differences between fission and fusion", response_model=KeyDifferences)

Conditional workflows

Use a handler that decides to CONTINUE or STOP.

from pydantic import BaseModel
class Triage(BaseModel):
  ambiguous: bool
  follow_up_details: str
  reply_to_user: str

def triage_handler(output: Triage):
  return ("STOP", None) if output.ambiguous else ("CONTINUE", output.follow_up_details)

wf = (
  client.Workflow()
  .add(researcher, response_model=Triage, handler=triage_handler)
  .add(summarizer)
)