Hierarchical Agent Teams: Boosting Reliability in Agentic AI

The fifth installment of the "Building the 14 Key Pillars of Agentic AI" series focuses on the hierarchical agent‑group pattern, a reliability‑enhancing design that decomposes complex tasks into a planner (orchestrator) and a set of specialist executor agents.

Key design patterns for reliable agentic systems

Parallel tools : agents issue independent API calls in parallel to hide I/O latency.

Hierarchical agents : a manager splits a task into smaller 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 mixed retrieval : multiple retrieval strategies run together to improve context quality.

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

Complex tasks often introduce unexpected delays between planning and action. The hierarchical approach solves this by (1) assigning the overall task to an orchestrator that only plans, (2) delegating sub‑tasks to specialist executor agents, and (3) having the orchestrator synthesize the executors' results into a final output.

Structured data contracts

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

class FinancialData(BaseModel):
    """Structured output for the financial‑analysis agent."""
    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):
    """Structured output for the news‑and‑market analysis agent."""
    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):
    """Final investment‑report output of the chief analyst."""
    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 with brief justification.")

These Pydantic models act as a formal contract between specialist agents and the final synthesizer. A TeamGraphState TypedDict aggregates the outputs:

from typing import TypedDict, Annotated, Optional, List

class TeamGraphState(TypedDict):
    company_symbol: str
    company_name: str
    financial_data: Optional[FinancialData]
    news_analysis: Optional[NewsAndMarketAnalysis]
    final_report: Optional[FinalReport]
    performance_log: Annotated[List[str], operator.add]

Specialist executor nodes

The financial‑analysis node creates a focused prompt, calls a single tool, and forces the result into the FinancialData 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‑analysis node follows the same pattern with its own prompt and tool set. The synthesizer (chief analyst) receives the structured outputs and produces a FinalReport:

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]}

Graph workflow with LangGraph

The agents are wired together using StateGraph:

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

Performance and quality comparison

The notebook prints a side‑by‑side comparison of a monolithic agent and the hierarchical team. Sample output shows the monolithic report and the JSON‑formatted hierarchical report. The measured times are:

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

Because the two specialist executors run in parallel (6.89 s and 8.12 s respectively), the parallel stage completes in the longest executor's time (8.12 s) instead of the sequential sum (≈15 s). The hierarchical report contains a detailed financial summary, market analysis, and a justified recommendation, whereas the monolithic report is shorter and less structured.

These results demonstrate that a decoupled, specialist‑orchestrator architecture not only improves reliability through clear data contracts but also reduces latency and produces higher‑quality output.

References

Building the 14 Key Pillars of Agentic AI – https://levelup.gitconnected.com/building-the-14-key-pillars-of-agentic-ai-229e50f65986

Speculative execution – https://en.wikipedia.org/wiki/Speculative_execution

Redundant execution – https://developer.arm.com/community/arm-community-blogs/b/embedded-and-microcontrollers-blog/posts/comparing-lock-step-redundant-execution-versus-split-lock-technologies

🤖 Agentic Parallelism: A Practical Guide 🚀 – https://github.com/FareedKhan-dev/agentic-parallelism