Hierarchical Agent Groups: Boosting Reliability in Agentic AI

The series introduces design patterns that enhance the reliability of modern agentic AI systems. This fifth article focuses on the hierarchical agent‑group pattern, where a high‑level orchestrator (or manager) breaks complex tasks into smaller sub‑tasks delegated to specialized executor agents.

Key Reliability Patterns

Parallel Tools : agents run independent API calls in parallel to hide I/O latency.

Hierarchical Agents : a manager splits tasks into steps handled by executor agents.

Competitive Agent Ensembles : multiple agents propose answers and the system selects the best.

Redundant Execution : two or more agents solve the same task to detect errors.

Parallel and Hybrid Retrieval : multiple retrieval strategies run together to improve context quality.

Multi‑hop Retrieval : agents iteratively retrieve deeper, more relevant information.

The article then compares a monolithic agent with a hierarchical team on an investment‑report generation task, demonstrating that the hierarchical approach is faster and produces higher‑quality, more detailed reports.

Structured Communication Model

A structured data model is defined to bind the agents together. The following Pydantic models are used:

from langchain_core.pydantic_v1 import BaseModel, Field
from typing import Optional, List

class FinancialData(BaseModel):
    """Financial analysis agent's structured output."""
    price: float = Field(description="Current stock price.")
    market_cap: int = Field(description="Total market capitalization.")
    pe_ratio: float = Field(description="Price-to-Earnings ratio.")
    volume: int = Field(description="Average trading volume.")

class NewsAndMarketAnalysis(BaseModel):
    """News and market analysis agent's structured output."""
    summary: str = Field(description="A concise summary of the most important recent news and market trends.")
    competitors: List[str] = Field(description="A list of the company's main competitors.")

class FinalReport(BaseModel):
    """Chief analyst's final investment report."""
    company_name: str = Field(description="The name of the company.")
    financial_summary: str = Field(description="A paragraph summarizing the key financial data.")
    news_and_market_summary: str = Field(description="A paragraph summarizing the news, market trends, and competitive landscape.")
    recommendation: str = Field(description="A final investment recommendation (e.g., 'Strong Buy', 'Hold', 'Sell') with a brief justification.")

These models act as contracts that guarantee each specialist agent returns well‑defined data, which the orchestrator can reliably consume.

Specialist Executor Agents

Two executor agents are defined: a financial analyst and a news analyst. Each agent uses a narrowly focused prompt and a single tool, then forces its output into the corresponding Pydantic model.

from langchain.agents import create_tool_calling_agent, AgentExecutor
import time

financial_analyst_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert financial analyst. Your sole job is to use the provided tool to get key financial metrics for a company and return them in a structured format."),
    ("human", "Get the financial data for the company with stock symbol: {symbol}")
])

financial_agent = create_tool_calling_agent(llm, [get_financial_data], financial_analyst_prompt)
financial_executor = AgentExecutor(agent=financial_agent, tools=[get_financial_data]) | llm.with_structured_output(FinancialData)

def financial_analyst_node(state: TeamGraphState):
    print("--- [Financial Analyst] Starting analysis... ---")
    start_time = time.time()
    result = financial_executor.invoke({"symbol": state['company_symbol']})
    execution_time = time.time() - start_time
    log = f"[Financial Analyst] Completed in {execution_time:.2f}s."
    print(log)
    return {"financial_data": result, "performance_log": [log]}

The news analyst node follows the same pattern with its own prompt and tool.

Orchestrator (Report Synthesizer) Agent

report_synthesizer_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are the Chief Investment Analyst. Your job is to synthesize the structured financial data and market analysis provided by your specialist team into a final, comprehensive investment report, including a justified recommendation."),
    ("human", "Please create the final report for {company_name}.

Financial Data:
{financial_data}

News and Market Analysis:
{news_analysis}")
])

synthesizer_chain = report_synthesizer_prompt | llm.with_structured_output(FinalReport)

def report_synthesizer_node(state: TeamGraphState):
    print("--- [Chief Analyst] Synthesizing final report... ---")
    start_time = time.time()
    report = synthesizer_chain.invoke({
        "company_name": state['company_name'],
        "financial_data": state['financial_data'].json(),
        "news_analysis": state['news_analysis'].json()
    })
    execution_time = time.time() - start_time
    log = f"[Chief Analyst] Completed report in {execution_time:.2f}s."
    print(log)
    return {"final_report": report, "performance_log": [log]}

Workflow Construction with LangGraph

from langgraph.graph import StateGraph, END

workflow = StateGraph(TeamGraphState)
workflow.add_node("financial_analyst", financial_analyst_node)
workflow.add_node("news_analyst", news_analyst_node)
workflow.add_node("report_synthesizer", report_synthesizer_node)

workflow.set_entry_point(["financial_analyst", "news_analyst"])
workflow.add_edge(["financial_analyst", "news_analyst"], "report_synthesizer")
workflow.add_edge("report_synthesizer", END)

app = workflow.compile()

inputs = {"company_symbol": "TSLA", "company_name": "Tesla", "performance_log": []}
start_time = time.time()
team_result = None
for output in app.stream(inputs, stream_mode="values"):
    team_result = output
end_time = time.time()
team_time = end_time - start_time

The graph runs the two specialist agents in parallel, waits for both to finish, then invokes the synthesizer.

Performance and Quality Comparison

The article prints a side‑by‑side comparison of the monolithic agent report and the hierarchical team report, then shows timing results:

Monolithic Agent Total Time: 18.34 seconds
Hierarchical Team Total Time: 13.57 seconds
Time Saved: 4.77 seconds (26% faster)

Financial Analyst time: 6.89 s
News Analyst time: 8.12 s
Parallel stage time (max worker): 8.12 s
Sequential stage time (sum): 15.01 s

The hierarchical system produces a more detailed final report (including explicit financial figures, market summary, and a justified recommendation) and completes the parallel stage 26 % faster because the two specialist agents run concurrently.

Conclusion

Specializing tasks and decoupling them via a hierarchical orchestrator yields higher‑quality, more reliable outputs while reducing overall latency. Structured Pydantic contracts and parallel execution are key enablers of these gains.