Decentralized Blackboard Collaboration: A High‑Reliability Design Pattern for Agentic AI

Reliability patterns for agentic AI

Common patterns include parallel tool use, hierarchical agents, competitive agent ensembles, redundant execution, parallel and hybrid retrieval, and multi‑hop retrieval.

Decentralized Blackboard Collaboration

The pattern consists of a shared data space (the “blackboard”) and a set of independent specialist agents that continuously monitor the board. An agent is activated opportunistically when the blackboard state matches its expertise, reads the current state, writes its contribution, and returns to sleep. This yields a dynamic, emergent workflow where the solution is assembled piece‑by‑piece by the most relevant expert at each stage.

Example: Customer‑Support Ticket Processing

A three‑agent pipeline (analyzer, retriever, draftsman) processes a support ticket. Structured data objects published to the blackboard are defined with Pydantic models:

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

class ProblemAnalysis(BaseModel):
    """Structure the analysis of a user's problem, published by the analyzer agent"""
    product: str = Field(description="The product the user is having an issue with.")
    problem_summary: str = Field(description="A concise, one-sentence summary of the technical problem.")
    user_sentiment: Literal["Positive", "Negative", "Neutral"] = Field(description="The user's sentiment.")

class Solution(BaseModel):
    """Potential solutions published by the retriever agent"""
    relevant_articles: List[str] = Field(description="A list of knowledge base articles relevant to the problem.")

class DraftResponse(BaseModel):
    """Final reply drafted by the draftsman agent"""
    response_text: str = Field(description="The complete, user‑facing response drafted by the agent.")

The blackboard state type aggregates the ticket and optional slots for each agent’s output:

from typing import TypedDict, Annotated, Optional, List

class BlackboardState(TypedDict):
    ticket: str
    analysis: Optional[ProblemAnalysis]
    solution: Optional[Solution]
    draft: Optional[DraftResponse]
    performance_log: Annotated[List[str], lambda a, b: a + b]

Analyzer node reads the ticket, invokes a language model with a structured prompt, logs execution time, and publishes a ProblemAnalysis object:

from langchain_core.prompts import ChatPromptTemplate
import time

analyzer_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a Problem Analyzer. Your job is to read a customer support ticket, identify the product, summarize the problem, and gauge the user's sentiment."),
    ("human", "Please analyze the following ticket:

---
{ticket}
---")
])

analyzer_chain = analyzer_prompt | llm.with_structured_output(ProblemAnalysis)

def analyzer_node(state: BlackboardState):
    """First activated agent: read the ticket and publish analysis to the blackboard"""
    print("--- [AGENT: Problem Analyzer] Activating... ---")
    start_time = time.time()
    result = analyzer_chain.invoke({"ticket": state['ticket']})
    execution_time = time.time() - start_time
    log = f"[Analyzer] Completed in {execution_time:.2f}s."
    print(log)
    return {"analysis": result, "performance_log": [log]}

Other agents ( retriever_node and draftsman_node) follow the same pattern: they read the blackboard, perform their specialized work, and write to the solution or draft slots.

Central router inspects the blackboard after each node and decides the next agent:

def router(state: BlackboardState) -> str:
    """Inspect the blackboard and decide the next agent"""
    print("--- [ROUTER] Inspecting blackboard... ---")
    if state.get('draft'):
        print("--- [ROUTER] Decision: Draft is complete. Finishing workflow. ---")
        return END
    if state.get('solution'):
        print("--- [ROUTER] Decision: Solution found. Activating Draftsman. ---")
        return "draftsman"
    if state.get('analysis'):
        print("--- [ROUTER] Decision: Analysis complete. Activating Solution Retriever. ---")
        return "retriever"
    return "analyzer"

The workflow is assembled with StateGraph, creating a hub‑and‑spoke architecture where each specialist node routes back to the router:

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(BlackboardState)
workflow.add_node("analyzer", analyzer_node)
workflow.add_node("retriever", retriever_node)   # defined elsewhere in the notebook
workflow.add_node("draftsman", draftsman_node)   # defined elsewhere in the notebook
workflow.add_edge(START, "analyzer")
workflow.add_conditional_edges("analyzer", router)
workflow.add_conditional_edges("retriever", router)
workflow.add_conditional_edges("draftsman", router)
app = workflow.compile()
print("Graph constructed and compiled successfully.")

Result and analysis

Running the graph on a sample ticket yields a final blackboard state containing the original ticket, a structured analysis, retrieved articles, and a drafted response. Sample performance logs:

[Analyzer] Completed in 4.55s.

[Retriever] Completed in 7.89s.

[Draftsman] Completed in 6.21s.

Three key advantages observed:

Decoupling reduces errors : separating analysis from retrieval ensures the retrieval step operates on a clear, structured summary, lowering the chance of misinterpretation.

Specialization deepens quality : the draftsman receives sentiment, summary, and solution data, enabling it to craft a more helpful and empathetic reply.

Auditability and modularity : each blackboard object ( analysis, solution, draft) is an independent artifact, making debugging and incremental improvement easier than with a monolithic agent.

All theory and code are available in the GitHub repository:

https://github.com/FareedKhan-dev/agentic-parallelism

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