-: Hey, welcome, and in this lesson,
we're gonna have a look at orchestrator workers.
This is basically a common principle
where you have an LLM which acts as the orchestrator,
and that LLM will then decide to generate specific types
of LLM calls or specific types of tasks.
We then have those that get synthesized,
and then we have an output.
Basically, you can think of this like the LLM
will break down the tasks into various subtasks.
These are dynamically determined based on the input.
These are then executed in parallel,
and then the orchestrator synthesizes those
into the final results.
Some use cases are breaking down a coding problem
into subtasks,
using an LLM to generate code for each subtask,
and then making sure that that entire solution works,
or it could be, for example, dividing a data analysis task
into things like cleaning the data, identifying the trends,
and generating the visualizations.
Each step is separately handled by a separate work LLM,
and the orchestrator, basically,
will just integrate their findings
into the complete analytical report.
Okay, so the first thing we're gonna do
is we're gonna create two Pydantic models,
one called subtask, which will have the name
and description of the specific type of keys,
so you can see these properties here.
We've got the name, which is the type of string,
and that has a description of the name of the subtask.
We also have the description colon string,
and that is also a required field.
Notice the triple dots,
and then we're gonna have a brief description
of the subtask.
We're also going to create a separate Pydantic model
called orchestrator output, and that is also a base model,
and inside of that, we will have both an objective key,
which is a type of string and is a required field,
then we will have a description,
and then a summary of the coding task.
We are specifically making an orchestrator for coding.
We will also have a subtask property,
which will have a list of subtasks,
and then that will be a required field.
Notice the triple dots again,
and then we'll also give this a description
and say that this is a list of subtasks
to solve the coding task.
Okay, cool, now that we have that,
we're gonna create three separate prompts,
so the first prompt that we're gonna make
is called an orchestrator prompt,
and that's basically gonna break down our task
into a variety of, "You are a skilled software engineer.
Read the coding problem and break it down
into subtasks in JSON format.
Summarize the objective of the task,
and then list the subtasks to solve the coding task.
Provide your answer in JSON format with these fields."
The second one that we're gonna make is the worker prompt,
and for the worker prompt, we're basically gonna tell it,
"You are a skilled software engineer.
Read the subtask and generate code to solve the subtask,"
and we've got here subtask name and the subtask description
with those in there as well.
We'll also tell it, "Return only the code.
Make sure that the code is valid Python code."
Then for the final prompt, this is the aggregator prompt,
and this prompt here is specifically going to be useful
for synthesizing all of those subtasks
that have been completed into a final aggregation source,
so, "You are an experienced integrator of code.
We have code snippets from different subtasks.
Your job is to integrate the code snippets
into a complete solution,"
and we've given it the subtasks' code, and we say,
"Return only the complete code.
Do not include any other text or comments."
So we have our three prompts,
and then we're gonna need three different functions
to call each of those individual prompts,
so we're gonna have the first one,
which is a asynchronous def call orchestrator,
which will have a problem and a model,
and it will return a orchestrator output.
Now, the easy thing that we can do is just say,
prompt is equal to orchestrator prompt dot,
and we'll just use the format,
and then we will also then create a ChatGPT call here,
but we will use the client.beta.chat.completions.parse,
and then what we will put in here is we'll put the model,
we'll put the messages,
and we will also put the response format,
so you can see here,
and then we have done some simple validation,
if not response.choices, square bracket, 0.message.content,
or response.choices.message.parse is not available there,
then we raise an error.
Else, we just get the parsed orchestrator output out.
So you'll see that here is the type
of orchestrator output when you hover
over the dot parse in VSCode or cursor or some other IDE.
Okay, cool, so we've done the orchestrator call.
The next one we're gonna want is the def call worker,
which will basically take a name,
it will take a description of a task,
and it will also have a model,
which in this case, is our model, and it returns a string.
Now, in this scenario, we're just getting the worker prompt,
giving it the name of the task, the description of the task,
and also then making a chat completions call,
but we aren't using the beta client
because we don't need a responsive format out
from structured outputs,
so we're just using the standard client,
and we're returning either a string or none type.
Now, the function that we need
is the call aggregator function,
which will take the subtasks' code,
and it'll take a model as the arguments.
It will format the aggregated prompt with the subtask code,
it will make a call to ChatGPT,
and it will return the message.
Okay, now that we have all of these functions,
the next thing that we're gonna need to do
is define a workflow for this,
so we'll say def call orchestrator,
and then basically, what this is gonna do
is it's gonna call all of the orchestrator flow
with the problem, and the model is gonna be passed in,
and then the first thing we're gonna do,
so we'll just break this down, is we'll say step one,
the orchestrator breaks the main problem into subtasks,
step two, we have the parallel workers work
and handle each subtask,
and then we aggregate those results,
and then you'll see the next bit here,
we've got an entry point for this,
and let me just change this,
so we don't want call orchestrator,
we actually want this to be orchestrator workers flow,
and then let's also go and change
and put this code out inside of a new cell,
and let's give this a run,
and okay, so we've run into one error,
and what I've noticed with this error
is to make sure there isn't a comment after this model,
'cause that was causing an error,
and then after that, then you can run your code,
and so what's happening is we call the orchestrator,
we made it to this point here,
and then after that, we will have the orchestrator output,
which will have a bunch of subtasks,
and then each of those subtasks
will get executed in parallel,
and then we aggregate those solutions
so that we have the final code.
So if we go and have a look at this,
you can see this is the code that it decided to create,
and then it's put everything together all in one block.
So again, the whole point of this
is that the orchestrator is responsible
for breaking down the problem
into a series of orchestrator-outputted subtasks.
Those subtasks will then be executed
in a different LLM worker.
This LLM could have access to tools
to be able to deal with those various subtasks,
and then once all of those subtasks completed,
you'll call the aggregator
to then aggregate all those tasks.
There are other different types of frameworks
that attempt to try and do this, so BabyAGI
was one of the original frameworks
which allowed for task creation and then execution of tasks,
so that's originally where one of these ideas
for the orchestrator came about.
All right, cool. See you in the next video.