In the second part, we explored the principles of modular design. We discussed strategies for decomposing the Agent system by borrowing the bounded context concept from microservices to determine the scope of each sub-Agent.
We also hinted that modularization introduces a need for a well-thought-out interaction model between agents and sub-Agents.
Today, we will delve into the request dispatch pattern, which can help create a predictable mechanism for dispatching requests to sub-Agents and allowing these Agents to feedback results to the dispatcher.
Unified Dispatch/Callback Mechanism
When multiple Agents need to coordinate work within the agent system, you may create a series of ad-hoc calls and mismatched data structures. By standardizing how each Agent calls (or dispatches to) other Agents and how these Agents respond, you can reduce confusion, errors, and maintenance workload. A consistent interface forces each Agent to use the same “language” when making requests or returning results.
The motivation for a unified interface stems from the reality that in a complex Agent system, a single Agent rarely handles all aspects of a user request. Users might ask questions like tracking a package, initiating a return, and checking warranty status in the same conversation. If your system simply delegates to sub-Agents chosen by a large language model (LLM), you need a unified way to pass request data and retrieve structured responses. By viewing these Agent handoffs as function calls with strict patterns, you can ensure that each Agent, whether a parent or sub-Agent, exchanges information in a predictable manner.
Without this uniformity, a parent Agent might expect one data representation while a sub-Agent returns something entirely different. Or you might find mismatches when one sub-Agent attempts to call another. Each small inconsistency can trigger confusing errors that are hard to debug, especially under the dynamic behavior of LLM-driven systems. Consistency in data shapes and parameter names is key to enabling LLMs to reliably infer which function to call and what data must be provided.
Python Example
The parent Agent needs to know how to correctly delegate tasks to each sub-Agent. You achieve this by exposing functions responsible for specific domains (in the sense of function calls to LLM). For example:
tools = [ { "type": "function", "function": { "name": "handoff_to_OrdersAgent", "description": "Handles order-related queries such as tracking or managing orders.", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "The unique ID of the user."}, "message": {"type": "string", "description": "The user's query."} }, "required": ["user_id", "message"] } } }, { "type": "function", "function": { "name": "handoff_to_ReturnsAgent", "description": "Handles return-related tasks, such as authorizing or tracking a return.", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "The unique ID of the user."}, "message": {"type": "string", "description": "The user's query."} }, "required": ["user_id", "message"] } } }]When the large language model (LLM) determines that an Agent is needed for order-related issues, it can call <span>handoff_to_OrdersAgent</span> with the necessary parameters. Then the parent Agent dispatches the request accordingly:
def dispatch_request(self, function_call): fn_name = function_call["name"] arguments = json.loads(function_call["arguments"]) if fn_name == "handoff_to_OrdersAgent": result = self.child_agents["OrdersAgent"].process_request(arguments) elif fn_name == "handoff_to_ReturnsAgent": result = self.child_agents["ReturnsAgent"].process_request(arguments) else: result = {"status": "error", "message": f"Unknown function {fn_name}"} return resultThis approach allows the parent Agent to focus on routing while each sub-Agent concentrates on its specific domain (orders, returns, product issues, etc.).
Within a sub-Agent, you can define functions related to its specific tasks. For instance, <span>OrdersAgent</span> might expose <span>lookupOrder</span> or <span>searchOrders</span> functions. The reasoning loop of the sub-Agent itself is confined to that domain, helping to avoid confusion and large prompt contexts.
class OrdersAgent: def __init__(self): self.functions = [ { "type": "function", "function": { "name": "lookupOrder", "parameters": { "type": "object", "properties": { "order_id": {"type": "string", "description": "The order ID."} }, "required": ["order_id"] } } }, { "type": "function", "function": { "name": "searchOrders", "parameters": { "type": "object", "properties": { "customer_id": {"type": "string", "description": "The customer ID."} }, "required": ["customer_id"] } } } ] def process_request(self, payload): self.message_history.append({"role": "user", "content": payload["message"]}) for _ in range(3): # Limit recursive calls response = self.run_llm_cycle(self.functions) if "function_call" in response: function_call = response["function_call"] result = self.handle_function_call(function_call) if result["status"] == "success": return result elif result["status"] == "escalate": return {"status": "escalate", "message": result["message"]} else: return {"status": "success", "data": response["content"]} return {"status": "error", "message": "Exceeded reasoning steps"} def handle_function_call(self, function_call): if function_call["name"] == "lookupOrder": return {"status": "success", "data": "Order details found..."} elif function_call["name"] == "searchOrders": return {"status": "success", "data": "Searching orders..."} else: return {"status": "escalate", "message": f"Function {function_call['name']} not supported"}Once the sub-Agent completes its task, it sends the results back to the parent Agent in a consistent format. This is the callback. The parent Agent can then:
-
If all goes well, pass the response back to the user.
-
Retry the request using another sub-Agent.
-
If the system cannot handle it automatically, escalate the issue to a human Agent.
For example:
response = orders_agent.handle_request(payload)if response["status"] == "success": parent_agent.add_message(role="assistant", content=response["data"])elif response["status"] == "escalate": parent_agent.add_message(role="system", content="OrdersAgent could not complete the request.") # Optionally retry with another agentIn any real-world system, certain queries may fail for some unforeseen reasons—like API downtime, data missing, or unsupported functions by the sub-Agent. When this happens, the sub-Agent returns an “escalate” status:
def handle_function_call(self, function_call): if function_call["name"] == "unsupported_function": return {"status": "escalate", "message": "Unsupported function"}The parent Agent can capture this status and decide whether to retry, escalate to another Agent, or ultimately return an error message to the user.
Looking Ahead
Between the second and third parts, we can see how the Agent system can be decomposed into a series of Agents/sub-Agents with a unified communication model to coordinate interactions throughout the Agent hierarchy.
However, these Agents do not exist in isolation. They need access to external tools, especially data. In the fourth part, we will explore the nuances of data retrieval in Agent systems and examine the unique data needs of Agent systems.