Orchestrator workflow in Langgraph
The Orchestrator -worker architecture
As a manager in your company, you are responsible for organizing, coordinating, and managing complex tasks to ensure that they function cohesively and are completed successfully. To accomplish this, you will divide your work into sections — such as Section 1, Section 2, and Section 3 — and assign these sections to your team members. For example, you might assign Section 1 and Section 3 to Worker A and Section 2 to Worker B.
All workers will perform their tasks simultaneously, and you will dynamically delegate work as needed. Once a worker completes their task, they will wait idly until the next assignment is given. After all tasks are completed, you will combine their outputs successfully — essentially synthesizing their work into a final product.
This process can be likened to an orchestrator-worker architecture, which enhances flexibility by breaking tasks down into subtasks that can be worked on in parallel. Each worker focuses on their
subtasks independently, allowing for efficient completion, and you will then synthesize their outputs into a cohesive final result.
The Orchestrator: The Manager who plans and breaks down the “multifaceted” goal.
Delegation: Assign sub-tasks to the right workers.
Parallelization: allows Workers A, B, and C to work simultaneously, saving time.
The Idle State: Workers wait quietly once their specific job is done.
Synthesis: The Manager’s final “glue” that creates a cohesive result or output.
Simple Orchestrator and Orchestrator–Worker :
Let’s understand the difference between Simple Orchestrator and Orchestrator–Worker.
A Simple Orchestrator : Let’s consider a real-life example regarding school admissions. When students apply for school admission, their application must be reviewed by multiple departments, so the administration divides the tasks into four categories:
📚 Academic Department
🌍 Visa Check
🏥 Healthcare Department
💰 Finance Department
Here, an admission coordinator or administrator can decide which department runs next.
So the Langraph concept is normal for this scenario.
An Orchestrator–Worker : Let’s consider a real-life example regarding academic evaluation of multiple subjects such as English, Mathematics, Science (Biology, Chemistry, Physics), Humanities (History, Geography, Economics), and Languages, Technologies, and Physical Education. So we don’t know how many subjects students submit, how many subject tasks there are, and based on this, we can evaluate their final results.
Let’s understand this example :
STEP 1 :Define Structured Output for Planning: A schema for organized output to utilize in planning activities
# Individual Section (Task)
from pydantic import BaseModel, Field
from typing import List
class Section(BaseModel):
"""
One evaluation task for a subject.
"""
name: str = Field(description="Task name, e.g., subject to evaluate")
description: str = Field(description="Instructions for this evaluation")
# Collection of Sections (Task)
class Sections(BaseModel):
sections: List[Section] = Field(
description="List of dynamic evaluation tasks for subjects"
)
# Augment the LLM with schema for structured output
planner = llm.with_structured_output(Sections)
STEP 2 : Expand Graph State
from typing import TypedDict, List,Annotated
from langgraph.types import Send import operator
# Graph state
class State(TypedDict):
student_id: str
subjects : List[str] # List of transcripts to evaluate
sections: List[Section] # Planned evaluation tasks
completed_evaluations: Annotated[List[str], operator.add]
# combined Worker outputs collected
final_report: str
STEP 3 : A worker state Each worker receives their own assigned task, which is subject to evaluation.
# Worker state
class WorkerState(TypedDict):
section: Section
completed_evaluations: List[str]
# list of outputs; each worker appends their results here.
STEP 4 : Define Nodes
Node 1 : An Orchestrator Node: Create a section and delegate to this worker using the send API, which dynamically generates a worker node.
def orchestrator(state: State)-> List[Section]:
"""
Generate dynamic sections (tasks) from subjects.
"""
subjects = state["subjects"]
sections = [
Section(
name=subject,
description=f"Evaluate performance in {subject}"
)
for subject in subjects
]
return {"sections": sections}
Node 2: Worker Node : Runs one section or subject at a time.
def llm_call_to_evaluate_subject(state: WorkerState):
""" Each worker evaluates a single assigned section (subject)
and returns the evaluation."""
section = state["section"]
# Simulate evaluation — replace with actual LLM call if needed
eval_summary = f"Evaluated {section.name}.Done 👍"
# Return result to shared state
return {"completed_evaluations": [eval_summary]}
Each worker has its own state, represented by the class `WorkerState`, and all worker outputs are written to a shared state key. This key is accessible to the orchestrator graph, allowing the Orchestrator to access all worker outputs and synthesize them into a final output using the Synthesizer Node.
Node 3 : Synthesizer Node
def synthesizer(state: State):
"""
Synthesize all completed evaluations into a final report.
"""
completed = state.get("completed_evaluations", [])
combined_report = "\n\n".join(completed)
return {"final_report": f"Final academic evaluation:\n{combined_report}"}
Conditional Edge : A function to create llm_call workers that each evaluate.
def assign_workers(state: State):
"""
Assign a worker to each section using Send API.
"""
sections = state.get("sections", [])
# Dynamically create worker nodes using Send()
return [Send("llm_call_to_evaluate_subject", {"section": s})
for s in sections]
A graph-based agent orchestration framework allows you to connect nodes—each representing actions—via edges that determine the task flow. These edges keep the sequence, while conditional edges enable branching depending on decisions, creating a workflow that is flexible and adaptable.
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from IPython.display import display, Image, Markdown
STEP 5 : Create Builder
orchestrator_worker_builder = StateGraph(State) STEP 6: Add Nodes
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node(
"llm_call_to_evaluate_subject",
llm_call_to_evaluate_subject
)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)
STEP 7 : Add Edges
# Start → Orchestrator
orchestrator_worker_builder.add_edge(START, "orchestrator")
# Orchestrator → Worker (dynamic via Send)
orchestrator_worker_builder.add_conditional_edges(
"orchestrator",
assign_workers,
["llm_call_to_evaluate_subject"])
# Worker → Synthesizer
orchestrator_worker_builder.add_edge(
"llm_call_to_evaluate_subject",
"synthesizer"
)
# Synthesizer → End
orchestrator_worker_builder.add_edge("synthesizer", END)
STEP 8 : Compile Workflow
orchestrator_worker = orchestrator_worker_builder.compile()STEP 9 : Invoke Workflow
state = orchestrator_worker.invoke({
"student_id": "S123",
"subjects": ["Math", "Physics", "Chemistry"],
"sections": [],
"completed_evaluations": [],
"final_report": ""
})STEP 10 : Show Final Synthesised Report
display(Markdown(state["final_report"]))
Mermaid Image :Orchestrator–Worker Workflow in LangGraph
In summary :
Worker State: Each worker only sees the section they are assigned to, keeping the prompt clean.
Shared State: Employing operator.add guarantees the automatic aggregation of parallel outputs.
Scalability: Whether a student submits two subjects or twenty, the Send() API dynamically scales the workers.
Ultimately, the decision between a simple orchestrator and an orchestrator–worker depends on the nature of the task. The Orchestrator–Worker model dynamically divides the workload, allowing for massively parallel processing and handling an infinite amount of data, making it suitable for data processing, research, and analysis. In contrast, a simple orchestrator follows a fixed blueprint used for business workflows such as approvals and KYC, with known work.
References :
