Hi everyone:
Recently i was working on multiagent system with langrapgh and after completing, i though why not publish it as blog post to guide junior developer familarize with multiagent langraph
The workflow is relatively simple. There are 3 agents
search_agent: it's job is to search internet and collect upto 5 related search results with the help of DDGS library.
planning agent: gets information from search_agent and ask llm to make plaining strategy to implement execution
execution agent: takes information from planning agent and execute the required action like making startup project folders or installing libraries etc
orchestration controls the workflow and route which agent needs to take implement at current timestamp.
The overall workflow is like, User query--> orchestrator --> search_agent --> orchestrator --> planning agent --> orchestrator --> execution_agent --> orchestrator-- end
Understanding State in LangGraph
Before deep dive into the project soure code we need to understand what is state in project. In LangGraph, state is the shared data structure that flows through the graph. Agents can read information from the state and return updates that are then available to other nodes.
Here is the state class
class MultiAgentState(MessagesState): MultiAgentState inherits from MessagesState, which already provides the messages field used to store the conversation messages.
"""State shared across all agents - inherits messages from MessagesState"""
current_agent: str
current_model: str
gpu_memory: Dict[str, int]
context: Dict[str, Any]
task_queue: Annotated[List[Dict[str, Any]], task_queue_reducer] # CRITICAL FIX: Use custom reducer
agent_states: Dict[str, Dict[str, Any]]
error_count: int
metadata: Dict[str, Any]
current_agent: identifies which agent or workflow node should run next
context: contain additional information like data fetch from search agent
task_queue: data from the end result of planing agent which need to be executed by execution agent
error_count: keep track of workflow errors
metadata: store additional execution information
Furthermore, lets dive into the project source code so we can understand what each part is doing.
1. orchestrator.py
This is the main file which controls which agent needs to run. First of all, i registered the three agents(search,planning and execution agent) with the help of OrchestratorAgent.register_agent(). This allows the orchestrator to route the workflow to those agents.
OrchestratorAgent.process() This is where the main routing is happening. First it check if the state contains any messaage, if no then workflow terminates. Next it checks if there are any pending task in the task_queue. If pending task exit then the next agent will be executor because the pending work needs to be perform first.if there are no pending task the orchestor will check whether web research is completed.
state.context.web_search_completed = False --> Search Agent. After search agent the state will be updated and state.context.web_search_completed = True. The orchestrotor sees the search_agent has done its job and looks whether the planning agent is being called state.context.planning_completed = False --> Planning Agent.The planning agent generate the executeable task and put them into state.task_queue_. At this point the orchestrator sees pending task and route execute agent next. The cycle repeats until there is no pending task in the state.task_queue. Finally: No pending tasks+ All workflow stages completed--> END.
2. Seach_agent.py
This is the simplest agent and it's work is to get user query, search internet using DDGS python library.
- PerplexicaSearchAgent.process(): receives the state and checks if there is any message in the state.message. if not it will be call orchestrator again saying no query found for the current message. next it will call DDGS with the user message and receives responce. After formating the responce, the function return updated states
return { "messages": [ AIMessage(content=response) ], "current_agent": "orchestrator", "context": { **state.get("context", {}), "search_results": results, "web_search_completed": True, "last_search": { "query": query, "results": results[:5], "timestamp": datetime.now().isoformat() } } }
3. planning_agent.py
The Planning Agent takes the user’s request and the web-search results as input.
It uses an LLM to break the request into clear, step-by-step technical tasks.
These tasks are then placed into the task queue for the Execution Agent to perform.
def init_llm(): For the LLM, I used openai/gpt-oss-120b through Groq. I chose Groq because it provides a fast API and has a free usage tier.
return ChatGroq(
model="openai/gpt-oss-120b",
temperature=0,
max_tokens=1536
)
In Planningagent.process() it first check if there is any pending task in the state.task_queue. if yes then it goes back to orchestrator with state.current_agent == "executor", returns "current_agent": (
"executor" if pending_tasks else "end"
),
Then it will get user original message and search result from state.context.get("search_results", []),. after getting the required data the function call
task = self._extract_task(state.get("messages", []))
plan = await self._generate_plan(. _generate_plain is actually calling the llm with prompt with relevent user message and searched result. The final return of Planningagent.process() is updated with "current_agent" =="execute",
task=task,
search_results=search_results
)
"context": {
**context,
"current_plan": plan,
"planning_completed": True,
"planning_timestamp": datetime.now().isoformat()
}
4. execution_agent.py
The Execution Agent receives the tasks generated by the Planning Agent.
It identifies the type of task and selects the appropriate execution method.
It performs actions such as installing dependencies, creating folders, and generating project files/code.
After completing each task, it updates the task queue and continues until all tasks are completed
it first check the state.task_queue if there is any pending task, so it will assign the given task to a variable current_task.
Again the function checks is any pending task is available. If not the the function calls self._handle_direct_execution(state)
In one sentence: _handle_direct_execution() is a fallback path that allows the Execution Agent to handle a user's direct execution request without waiting for the Planning Agent to create a task queue.
if there is pending task in the state.task_queue. it will call result = await self._execute_task(current_task). execut_task purpose is to get the task header, description and checks if there is any word in the header or description containing "install",
"dependencies",
"libraries",
"packages",
"virtual environment",. if yes return await self._execute_installation(task)is called which main purpose os installing.
second option is that if word in the header or description contains _ "initialize repository",
"initialize project",
"starter project",
"project structure",
"project layout",
"scaffold",
"scaffolding",
"development environment",_ then await self._execute_project_setup(task) execute which main purpose is to create folders for the given topic name.
Third option is if words in header or description contains "http client",
"request handling",
"http requests", it will call self._execute_http_client(task)It typically creates the necessary Python file and code for making web requests, handling errors, and respecting things like robots.txt and rate limits.
and there are others function like self._execute_command(task),await self._execute_storage(task),await self._execute_crawler(task) which has different purposes.
The final return of process() function is updated_task = {.
**current_task,
"status": "completed" if result["success"] else "failed",
"completed_at": datetime.now().isoformat(),
"result": result
},
return {
"messages": [response_message],
"current_agent": next_agent,
"task_queue": updated_task_queue,
"context": {
**state.get("context", {}),
"last_execution": result,
"execution_timestamp": datetime.now().isoformat()
}
}
5. main.py
Entry point of the project which imports MultiAgentOrchestrator from orchestrator.py and initailizing all the agents after. Registering all the agents in orchestrator and buidling the graph.
User need to run the main file like python main.py research "Research the best approach for building a Python news web scraper, identify the required libraries and challenges, create a step-by-step implementation plan, and set up a starter project structure." and all the steps will be printed after that.
Github link
Please read the readme.md file to start the project.
If you have any question feel free to ask: shahalmovid@gmail.com
Happy Coding:
Top comments (0)