マルチエージェントシステムの構成

エージェントを組み合わせるための3つのオーケストレーション基本要素。

SequentialAgent — エージェントを順序通りに実行

各エージェントの最終回答が次のエージェントの入力になります。

from orxhestra import SequentialAgent, LlmAgent

researcher = LlmAgent(name="researcher", model=model, instructions="Research the topic.")
writer = LlmAgent(name="writer", model=model, instructions="Write an article from the research.")

pipeline = SequentialAgent(
    name="pipeline",
    agents=[researcher, writer],
)

async for event in pipeline.astream("AI trends 2025"):
    if event.is_final_response():
        print(event.text)

ParallelAgent — エージェントを同時に実行

すべてのエージェントが同時に実行されます。各エージェントはブランチ分離を持つ派生コンテキストを取得します。

from orxhestra import ParallelAgent, LlmAgent

analyst_a = LlmAgent(name="market", model=model, instructions="Analyze market trends.")
analyst_b = LlmAgent(name="tech", model=model, instructions="Analyze tech trends.")

parallel = ParallelAgent(
    name="analysis",
    agents=[analyst_a, analyst_b],
)

LoopAgent — 完了まで繰り返す

escalate=True（exit_loop_tool経由）またはmax_iterationsに達するまでサブエージェントを繰り返します。

from orxhestra import LoopAgent, LlmAgent
from orxhestra.tools import exit_loop_tool

writer = LlmAgent(name="writer", model=model, instructions="Write a draft.")
reviewer = LlmAgent(
    name="reviewer",
    model=model,
    instructions="Review the draft. Call exit_loop if approved.",
    tools=[exit_loop_tool],
)

loop = LoopAgent(
    name="review_loop",
    agents=[writer, reviewer],
    max_iterations=5,
)

カスタム停止条件

def quality_check(ctx, last_event):
    score = ctx.state.get("quality_score", 0)
    return score >= 8  # stop if quality is high

loop = LoopAgent(
    name="quality_loop",
    agents=[writer, reviewer],
    should_continue=quality_check,  # return True to continue
    max_iterations=10,
)

パターンの組み合わせ

# Research in parallel, then write sequentially, then review in a loop
research = ParallelAgent(name="research", agents=[market_analyst, tech_analyst])
write = LlmAgent(name="writer", model=model, instructions="Synthesize research into article.")
review_loop = LoopAgent(name="review", agents=[editor, fact_checker], max_iterations=3)

full_pipeline = SequentialAgent(
    name="content_pipeline",
    agents=[research, write, review_loop],
)

トランスファールーティング — エージェントハンドオフ

from orxhestra import LlmAgent
from orxhestra.tools import make_transfer_tool

sales = LlmAgent(name="sales", model=model, description="Handles orders.", instructions="...")
support = LlmAgent(name="support", model=model, description="Technical help.", instructions="...")

triage = LlmAgent(
    name="triage",
    model=model,
    instructions="Route the user to the right specialist.",
    tools=[make_transfer_tool([sales, support])],
)
triage.register_sub_agent(sales)
triage.register_sub_agent(support)