From User Query to Execution: Building a Multi-Agent System with LangGraph

A developer published a walkthrough of a multi-agent system built with LangGraph that chains three agents — a search agent using the DDGS library, a planning agent, and an execution agent — under an orchestrator that routes control between them. The system shares a MultiAgentState object, inheriting from MessagesState, that carries fields including current_agent, context, task_queue, error_count, and metadata, with a custom reducer applied to the task queue. The orchestrator routes work by checking pending tasks and completion flags such as web_search_completed and planning_completed, looping until the queue is empty and all stages finish.

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 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. 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. self. execute command task , await self. execute storage task , await self. execute crawler task which has different purposes. 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 https://github.com/almovidhussaini/multi-agent-planning-websearch-execution Please read the readme.md file to start the project. If you have any question feel free to ask: shahalmovid@gmail.com mailto:shahalmovid@gmail.com Happy Coding: