all-agentic-architectures

name	all-agentic-architectures
description	Implementation guide for 17+ agentic AI architectures using LangChain and LangGraph for building sophisticated AI agents
triggers	["how do i build an agentic architecture","implement reflection pattern for ai agents","create multi-agent system with langgraph","set up tree of thoughts architecture","build react agent with tools","implement agent memory with episodic and semantic","create self-improving ai agent","design meta-controller for specialized agents"]

All Agentic Architectures Skill

Skill by ara.so — AI Agent Skills collection.

This skill provides comprehensive guidance for implementing 17+ state-of-the-art agentic architectures using LangChain and LangGraph. The project offers production-ready implementations of patterns ranging from simple reflection loops to complex multi-agent systems with memory, planning, and self-improvement capabilities.

What This Project Does

All Agentic Architectures is a comprehensive collection of modern AI agent design patterns implemented as runnable Jupyter notebooks. It covers:

• Single-Agent Patterns: Reflection, Tool Use, ReAct, Planning
• Multi-Agent Systems: Collaborative teams, Meta-Controllers, Blackboard systems, Ensemble patterns
• Advanced Memory: Episodic + Semantic memory, Graph-based world models
• Safety & Reliability: Dry-Run Harness, Plan-Execute-Verify, Simulators
• Self-Improvement: RLHF-style feedback loops, Metacognitive agents
• Complex Reasoning: Tree of Thoughts, Cellular Automata

Each architecture is designed for practical use across different stages of AI system development.

Installation

Basic Setup

# Clone the repository
git clone https://github.com/FareedKhan-dev/all-agentic-architectures.git
cd all-agentic-architectures

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: .\venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

Core Dependencies

pip install langchain langgraph langsmith pydantic
pip install openai anthropic  # For LLM providers
pip install tavily-python  # For search tool
pip install neo4j faiss-cpu  # For memory architectures
pip install jupyter notebook  # For running notebooks

Environment Variables

Create a .env file in the project root:

# LLM Provider (choose one or multiple)
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
NEBIUS_API_KEY=your_nebius_key

# Tools
TAVILY_API_KEY=your_tavily_key

# Memory Systems
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=your_neo4j_password

# LangSmith (optional, for tracing)
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your_langsmith_key
LANGCHAIN_PROJECT=agentic-architectures

Core Architecture Patterns

1. Reflection Pattern

The Reflection pattern creates a self-critiquing agent that iteratively improves its output.

from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
from pydantic import BaseModel
from typing import List, TypedDict

class ReflectionState(TypedDict):
    messages: List[HumanMessage | AIMessage]
    iterations: int

def generate_node(state: ReflectionState):
    """Generate initial response"""
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(model="gpt-4", temperature=0.7)
    response = llm.invoke(state["messages"])
    
    return {
        "messages": state["messages"] + [response],
        "iterations": state["iterations"]
    }

def reflect_node(state: ReflectionState):
    """Critique and improve the response"""
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(model="gpt-4", temperature=0.3)
    
    reflection_prompt = f"""Review the following response and provide constructive criticism:
    
Response: {state['messages'][-1].content}

Provide specific suggestions for improvement."""
    
    critique = llm.invoke([HumanMessage(content=reflection_prompt)])
    
    improvement_prompt = f"""Original task: {state['messages'][].content}

Previous response: 

Critique: 

Provide an improved response addressing the critique."""
    
    improved = llm.invoke([HumanMessage(content=improvement_prompt)])
    
     {
        : state[] + [critique, improved],
        : state[] + 
    }

 ():
    
     state[] >= :
         
     


workflow = StateGraph(ReflectionState)
workflow.add_node(, generate_node)
workflow.add_node(, reflect_node)

workflow.set_entry_point()
workflow.add_conditional_edges(
    ,
    should_continue,
    {: , : END}
)
workflow.add_conditional_edges(
    ,
    should_continue,
    {: , : END}
)

app = workflow.()


result = app.invoke({
    : [HumanMessage(content=)],
    : 
})

2. ReAct (Reasoning + Acting) Pattern

ReAct dynamically interleaves reasoning and tool use.

from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain import hub
from langchain_community.tools.tavily_search import TavilySearchResults

# Define tools
search = TavilySearchResults(max_results=3)

def calculator(expression: str) -> str:
    """Evaluate mathematical expressions"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {str(e)}"

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="Useful for searching current information on the internet"
    ),
    Tool(
        name="Calculator",
        func=calculator,
        description="Useful for mathematical calculations. Input should be a valid Python expression."
    )
]

# Create ReAct agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5
)

# Execute multi-step reasoning
result = agent_executor.invoke({
    "input": "What is the current population of Tokyo, and what is 15% of that number?"
})

3. Multi-Agent System

Specialized agents collaborate to solve complex tasks.

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class MultiAgentState(TypedDict):
    task: str
    research: str
    code: str
    review: str
    messages: Annotated[list, operator.add]

def research_agent(state: MultiAgentState):
    """Agent specialized in research"""
    from langchain_openai import ChatOpenAI
    from langchain_community.tools.tavily_search import TavilySearchResults
    
    llm = ChatOpenAI(model="gpt-4")
    search = TavilySearchResults()
    
    research_prompt = f"""Research the following task and provide comprehensive background:
    Task: {state['task']}
    
    Provide key technical details and best practices."""
    
    search_results = search.run(state['task'])
    response = llm.invoke(f"{research_prompt}\n\nSearch results: {search_results}")
    
    return {
        "research": response.content,
        "messages": [f"Research Agent: {response.content}"]
    }

def coding_agent(state: MultiAgentState):
    """Agent specialized in writing code"""
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(model="gpt-4", temperature=0.2)
    
    code_prompt = 
    
    response = llm.invoke(code_prompt)
    
     {
        : response.content,
        : []
    }

 ():
    
     langchain_openai  ChatOpenAI
    
    llm = ChatOpenAI(model=, temperature=)
    
    review_prompt = 
    
    response = llm.invoke(review_prompt)
    
     {
        : response.content,
        : []
    }


workflow = StateGraph(MultiAgentState)
workflow.add_node(, research_agent)
workflow.add_node(, coding_agent)
workflow.add_node(, review_agent)

workflow.set_entry_point()
workflow.add_edge(, )
workflow.add_edge(, )
workflow.add_edge(, END)

app = workflow.()


result = app.invoke({
    : ,
    : ,
    : ,
    : ,
    : []
})

4. Tree of Thoughts

Explore multiple reasoning paths systematically.

from typing import List, Dict, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class ThoughtNode(TypedDict):
    content: str
    score: float
    depth: int

class ToTState(TypedDict):
    problem: str
    thoughts: List[ThoughtNode]
    best_path: List[str]
    max_depth: int

def generate_thoughts(state: ToTState):
    """Generate multiple reasoning branches"""
    llm = ChatOpenAI(model="gpt-4", temperature=0.8)
    
    current_depth = max([t["depth"] for t in state["thoughts"]], default=0)
    
    # Get the best thoughts from current level
    current_thoughts = [t for t in state["thoughts"] if t["depth"] == current_depth]
    
    new_thoughts = []
    for thought in current_thoughts[:3]:  # Expand top 3 thoughts
        prompt = f"""Problem: {state['problem']}

Current reasoning: {thought[]}

Generate 3 different next steps or reasoning paths. Be creative and explore alternatives."""
        
        response = llm.invoke(prompt)
        
        
         i, line  (response.content.split()):
             line.strip():
                new_thoughts.append({
                    : thought[] +  + line.strip(),
                    : ,
                    : current_depth + 
                })
    
     {: state[] + new_thoughts}

 ():
    
    llm = ChatOpenAI(model=, temperature=)
    
    current_depth = ([t[]  t  state[]])
    current_thoughts = [t  t  state[]  t[] == current_depth]
    
    evaluated_thoughts = []
     thought  current_thoughts:
        eval_prompt = 
        
        response = llm.invoke(eval_prompt)
        :
            score = (response.content.strip())
        :
            score = 
        
        thought[] = score
        evaluated_thoughts.append(thought)
    
    
    all_thoughts = [t  t  state[]  t[] < current_depth] + evaluated_thoughts
    
     {: all_thoughts}

 ():
    
    current_depth = ([t[]  t  state[]], default=)
    
     current_depth >= state[]:
         
    
     

 ():
    
    
    max_depth = ([t[]  t  state[]])
    final_thoughts = [t  t  state[]  t[] == max_depth]
    
    best_thought = (final_thoughts, key= x: x[])
    
     {: best_thought[].split()}


workflow = StateGraph(ToTState)
workflow.add_node(, generate_thoughts)
workflow.add_node(, evaluate_thoughts)
workflow.add_node(, finalize_solution)

workflow.set_entry_point()
workflow.add_edge(, )
workflow.add_conditional_edges(
    ,
    should_continue,
    {: , : }
)
workflow.add_edge(, END)

app = workflow.()


result = app.invoke({
    : ,
    : [{
        : ,
        : ,
        : 
    }],
    : [],
    : 
})

5. Episodic + Semantic Memory

Combine conversational history with structured knowledge.

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_community.graphs import Neo4jGraph
from typing import List, Dict

class DualMemoryAgent:
    def __init__(self):
        # Episodic memory (vector store for conversations)
        self.embeddings = OpenAIEmbeddings()
        self.episodic_memory = FAISS.from_texts(
            ["Initial conversation"],
            self.embeddings
        )
        
        # Semantic memory (graph database for facts)
        self.semantic_memory = Neo4jGraph(
            url=os.getenv("NEO4J_URI"),
            username=os.getenv("NEO4J_USERNAME"),
            password=os.getenv("NEO4J_PASSWORD")
        )
        
        from langchain_openai import ChatOpenAI
        self.llm = ChatOpenAI(model="gpt-4")
    
    def add_episodic_memory(self, conversation: str):
        """Store conversation in vector database"""
        self.episodic_memory.add_texts([conversation])
    
    def add_semantic_fact(self, subject: str, relation: str, object: str):
        
        query = 
        .semantic_memory.query(query)
    
     () -> []:
        
        docs = .episodic_memory.similarity_search(query, k=k)
         [doc.page_content  doc  docs]
    
     () -> :
        
        query = 
        results = .semantic_memory.query(query)
        
        facts = []
         result  results:
            facts.append()
        
         .join(facts)
    
     () -> :
        
        
        past_conversations = .retrieve_episodic(user_input)
        
        
        entity_prompt = 
        entity_response = .llm.invoke(entity_prompt)
        entity = entity_response.content.strip()
        
        semantic_facts = .retrieve_semantic(entity)
        
        
        response_prompt = 
        
        response = .llm.invoke(response_prompt)
        
        
        .add_episodic_memory()
        
         response.content


agent = DualMemoryAgent()


agent.add_semantic_fact(, , )
agent.add_semantic_fact(, , )
agent.add_semantic_fact(, , )


response = agent.respond()

6. Meta-Controller Pattern

Route tasks to specialized sub-agents.

from enum import Enum
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class AgentType(str, Enum):
    RESEARCH = "research"
    CODING = "coding"
    WRITING = "writing"
    GENERAL = "general"

class MetaControllerState(TypedDict):
    user_input: str
    agent_type: AgentType
    final_response: str

def meta_controller(state: MetaControllerState):
    """Analyze task and route to appropriate specialist"""
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    routing_prompt = f"""Analyze this user request and classify it:

User request: {state['user_input']}

Categories:
- research: Questions requiring web search or current information
- coding: Programming, debugging, or technical implementation
- writing: Content creation, editing, or creative writing
- general: Casual conversation or simple questions

Respond with only one word: research, coding, writing, or general"""
    
    response = llm.invoke(routing_prompt)
    agent_type = response.content.strip().lower()
    
    return {"agent_type": AgentType(agent_type)}

def research_specialist(state: MetaControllerState):
    """Handle research-intensive queries"""
    from langchain_openai import ChatOpenAI
    from langchain_community.tools.tavily_search  TavilySearchResults
    
    llm = ChatOpenAI(model=)
    search = TavilySearchResults(max_results=)
    
    
    search_results = search.run(state[])
    
    
    synthesis_prompt = 
    
    response = llm.invoke(synthesis_prompt)
     {: response.content}

 ():
    
     langchain_openai  ChatOpenAI
    
    llm = ChatOpenAI(model=, temperature=)
    
    coding_prompt = 
    
    response = llm.invoke(coding_prompt)
     {: response.content}

 ():
    
     langchain_openai  ChatOpenAI
    
    llm = ChatOpenAI(model=, temperature=)
    
    writing_prompt = 
    
    response = llm.invoke(writing_prompt)
     {: response.content}

 ():
    
     langchain_openai  ChatOpenAI
    
    llm = ChatOpenAI(model=)
    response = llm.invoke(state[])
     {: response.content}

 () -> [, , , ]:
    
     state[].value


workflow = StateGraph(MetaControllerState)

workflow.add_node(, meta_controller)
workflow.add_node(, research_specialist)
workflow.add_node(, coding_specialist)
workflow.add_node(, writing_specialist)
workflow.add_node(, general_specialist)

workflow.set_entry_point()
workflow.add_conditional_edges(
    ,
    route_to_specialist,
    {
        : ,
        : ,
        : ,
        : 
    }
)

workflow.add_edge(, END)
workflow.add_edge(, END)
workflow.add_edge(, END)
workflow.add_edge(, END)

app = workflow.()


result = app.invoke({
    : ,
    : AgentType.GENERAL,
    : 
})

Configuration Patterns

LangSmith Tracing

Enable detailed tracing for debugging and monitoring:

import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGSMITH_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "my-agentic-project"

# All LangChain/LangGraph calls will now be traced

Custom LLM Configuration

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# Configure different models for different tasks
creative_llm = ChatOpenAI(
    model="gpt-4",
    temperature=0.9,
    max_tokens=2000
)

analytical_llm = ChatOpenAI(
    model="gpt-4",
    temperature=0.2,
    max_tokens=1000
)

# Use Anthropic for longer context
long_context_llm = ChatAnthropic(
    model="claude-3-opus-20240229",
    max_tokens=4096
)

Checkpointing for Long-Running Agents

from langgraph.checkpoint.sqlite import SqliteSaver

# Add persistence to any LangGraph workflow
memory = SqliteSaver.from_conn_string("checkpoints.db")

app = workflow.compile(checkpointer=memory)

# Run with thread_id for persistence
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(initial_state, config)

# Resume from checkpoint
continued = app.invoke(new_input, config)

Common Patterns and Best Practices

Pattern 1: LLM-as-a-Judge Evaluation

def evaluate_agent_output(task: str, output: str) -> dict:
    """Use LLM to evaluate agent performance"""
    from langchain_openai import ChatOpenAI
    
    judge_llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    eval_prompt = f"""Evaluate this AI agent output:

Task: {task}
Output: {output}

Rate on a scale of 1-10 for:
1. Correctness
2. Completeness
3. Clarity
4. Efficiency

Provide scores in JSON format:
{{"correctness": X, "completeness": X, "clarity": X, "efficiency": X, "reasoning": "..."}}
"""
    
    response = judge_llm.invoke(eval_prompt)
    
    import json
    return json.loads(response.content)

Pattern 2: Error Handling in Agents

from typing import TypedDict
from langgraph.graph import StateGraph, END

class RobustAgentState(TypedDict):
    input: str
    output: str
    error: str
    retry_count: int

def safe_agent_node(state: RobustAgentState):
    """Agent with error handling"""
    try:
        from langchain_openai import ChatOpenAI
        llm = ChatOpenAI(model="gpt-4", timeout=30)
        
        response = llm.invoke(state["input"])
        
        return {
            "output": response.content,
            "error": "",
            "retry_count": state["retry_count"]
        }
    
    except Exception as e:
        return {
            "output": "",
            "error": str(e),
            "retry_count": state["retry_count"] + 1
        }

def should_retry(state: RobustAgentState) -> str:
    """Decide whether to retry on error"""
    if state["error"] and state["retry_count"] < :
         
     state[]:
         
     

Pattern 3: Streaming Responses

from langchain_openai import ChatOpenAI

async def stream_agent_response(user_input: str):
    """Stream agent responses in real-time"""
    llm = ChatOpenAI(model="gpt-4", streaming=True)
    
    async for chunk in llm.astream(user_input):
        print(chunk.content, end="", flush=True)
        # Or yield chunk for web frameworks
        yield chunk.content

Running Notebooks

Start Jupyter

jupyter notebook

Recommended Notebook Order

1. Start with basics: 01_reflection.ipynb, 02_tool_use.ipynb, 03_ReAct.ipynb
2. Multi-agent fundamentals: 05_multi_agent.ipynb, 11_meta_controller.ipynb
3. Advanced memory: 08_episodic_with_semantic.ipynb, 12_graph.ipynb
4. Safety patterns: 06_PEV.ipynb, 14_dry_run.ipynb, 17_reflexive_metacognitive.ipynb
5. Complex reasoning: 09_tree_of_thoughts.ipynb, 10_mental_loop.ipynb

Execute Programmatically

import nbformat
from nbconvert.preprocessors import ExecutePreprocessor

def run_notebook(notebook_path: str):
    """Execute a notebook programmatically"""
    with open(notebook_path) as f:
        nb = nbformat.read(f, as_version=4)
    
    ep = ExecutePreprocessor(timeout=600, kernel_name='python3')
    ep.preprocess(nb, {'metadata': {'path': './'}})
    
    return nb

Troubleshooting

API Rate Limits

from langchain_openai import ChatOpenAI
from langchain.callbacks import get_openai_callback

# Track token usage
with get_openai_callback() as cb:
    response = llm.invoke("Your query")
    print(f"Tokens used: {cb.total_tokens}")
    print(f"Cost: ${cb.total_cost}")

# Add retry logic
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def resilient_llm_call(prompt):
    llm = ChatOpenAI(model="gpt-4")
    return llm.invoke(prompt)

Memory Issues with Large Graphs

# Use streaming for large outputs
from langgraph.graph import StateGraph

# Limit state retention
def trim_messages(state: dict, max_messages: int = 10):
    """Keep only recent messages"""
    if len(state.get("messages", [])) > max_messages:
        state["messages"] = state["messages"][-max_messages:]
    return state

Neo4j Connection Issues

from neo4j import GraphDatabase

def test_neo4j_connection():
    """Verify Neo4j connectivity"""
    try:
        driver = GraphDatabase.driver(
            os.getenv("NEO4J_URI"),
            auth=(
                os.getenv("NEO4J_USERNAME"),
                os.getenv("NEO4J_PASSWORD")
            )
        )
        
        with driver.session() as session:
            result = session.run("RETURN 1 AS test")
            print("✓ Neo4j connection successful")
            return True
    
    except Exception as e:
        print(f"✗ Neo4j connection failed: {e}")
        return False
    finally:
        driver.close()

Debugging LangGraph Workflows

from langgraph.graph import StateGraph

# Enable verbose output
workflow = StateGraph(StateType)
# ... add nodes ...

app = workflow.compile(debug=True)

# Visualize the graph
from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

# Step through execution
for step in app.stream(initial_state):
    print(f"Step: {step}")

Integration Examples

FastAPI Integration

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()