aradotso 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'][0].content}
Previous response: {state['messages'][-1].content}
Critique: {critique.content}
Provide an improved response addressing the critique."""
improved = llm.invoke([HumanMessage(content=improvement_prompt)])
return {
"messages": state["messages"] + [critique, improved],
"iterations": state["iterations"] + 1
}
def should_continue(state: ReflectionState):
"""Decide whether to continue reflection"""
if state["iterations"] >= 3:
return "end"
return "reflect"
# Build the graph
workflow = StateGraph(ReflectionState)
workflow.add_node("generate", generate_node)
workflow.add_node("reflect", reflect_node)
workflow.set_entry_point("generate")
workflow.add_conditional_edges(
"generate",
should_continue,
{"reflect": "reflect", "end": END}
)
workflow.add_conditional_edges(
"reflect",
should_continue,
{"reflect": "reflect", "end": END}
)
app = workflow.compile()
# Use the reflection agent
result = app.invoke({
"messages": [HumanMessage(content="Write a Python function to calculate Fibonacci numbers")],
"iterations": 0
})
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 = f"""Based on the research, implement the solution:
Task: {state['task']}
Research: {state['research']}
Provide production-ready, well-documented code."""
response = llm.invoke(code_prompt)
return {
"code": response.content,
"messages": [f"Coding Agent: {response.content}"]
}
def review_agent(state: MultiAgentState):
"""Agent specialized in code review"""
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4", temperature=0)
review_prompt = f"""Review the following code for quality, security, and best practices:
Task: {state['task']}
Code:
{state['code']}
Provide detailed feedback and suggestions."""
response = llm.invoke(review_prompt)
return {
"review": response.content,
"messages": [f"Review Agent: {response.content}"]
}
# Build multi-agent workflow
workflow = StateGraph(MultiAgentState)
workflow.add_node("research", research_agent)
workflow.add_node("code", coding_agent)
workflow.add_node("review", review_agent)
workflow.set_entry_point("research")
workflow.add_edge("research", "code")
workflow.add_edge("code", "review")
workflow.add_edge("review", END)
app = workflow.compile()
# Execute multi-agent collaboration
result = app.invoke({
"task": "Build a REST API rate limiter using Redis",
"research": "",
"code": "",
"review": "",
"messages": []
})
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['content']}
Generate 3 different next steps or reasoning paths. Be creative and explore alternatives."""
response = llm.invoke(prompt)
# Parse and create new thought nodes
for i, line in enumerate(response.content.split('\n\n')):
if line.strip():
new_thoughts.append({
"content": thought['content'] + " -> " + line.strip(),
"score": 0.0,
"depth": current_depth + 1
})
return {"thoughts": state["thoughts"] + new_thoughts}
def evaluate_thoughts(state: ToTState):
"""Score each thought based on quality"""
llm = ChatOpenAI(model="gpt-4", temperature=0.2)
current_depth = max([t["depth"] for t in state["thoughts"]])
current_thoughts = [t for t in state["thoughts"] if t["depth"] == current_depth]
evaluated_thoughts = []
for thought in current_thoughts:
eval_prompt = f"""Problem: {state['problem']}
Reasoning path: {thought['content']}
Rate this reasoning path from 0.0 to 1.0 based on:
- Logical soundness
- Progress toward solution
- Creativity
Respond with only a number."""
response = llm.invoke(eval_prompt)
try:
score = float(response.content.strip())
except:
score = 0.5
thought["score"] = score
evaluated_thoughts.append(thought)
# Keep all previous thoughts plus newly evaluated ones
all_thoughts = [t for t in state["thoughts"] if t["depth"] < current_depth] + evaluated_thoughts
return {"thoughts": all_thoughts}
def should_continue(state: ToTState):
"""Decide whether to continue exploring"""
current_depth = max([t["depth"] for t in state["thoughts"]], default=0)
if current_depth >= state["max_depth"]:
return "finalize"
return "generate"
def finalize_solution(state: ToTState):
"""Select and return the best reasoning path"""
# Find the best thought at maximum depth
max_depth = max([t["depth"] for t in state["thoughts"]])
final_thoughts = [t for t in state["thoughts"] if t["depth"] == max_depth]
best_thought = max(final_thoughts, key=lambda x: x["score"])
return {"best_path": best_thought["content"].split(" -> ")}
# Build ToT workflow
workflow = StateGraph(ToTState)
workflow.add_node("generate", generate_thoughts)
workflow.add_node("evaluate", evaluate_thoughts)
workflow.add_node("finalize", finalize_solution)
workflow.set_entry_point("generate")
workflow.add_edge("generate", "evaluate")
workflow.add_conditional_edges(
"evaluate",
should_continue,
{"generate": "generate", "finalize": "finalize"}
)
workflow.add_edge("finalize", END)
app = workflow.compile()
# Solve complex problem with ToT
result = app.invoke({
"problem": "Design a distributed caching system for a social media platform",
"thoughts": [{
"content": "Starting analysis",
"score": 1.0,
"depth": 0
}],
"best_path": [],
"max_depth": 3
})
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):
"""Store structured fact in graph database"""
query = f"""
MERGE (s:Entity {{name: '{subject}'}})
MERGE (o:Entity {{name: '{object}'}})
MERGE (s)-[r:{relation}]->(o)
"""
self.semantic_memory.query(query)
def retrieve_episodic(self, query: str, k: int = 3) -> List[str]:
"""Retrieve similar past conversations"""
docs = self.episodic_memory.similarity_search(query, k=k)
return [doc.page_content for doc in docs]
def retrieve_semantic(self, entity: str) -> str:
"""Retrieve facts about an entity"""
query = f"""
MATCH (e:Entity {{name: '{entity}'}})-[r]->(related)
RETURN type(r) as relation, related.name as object
LIMIT 10
"""
results = self.semantic_memory.query(query)
facts = []
for result in results:
facts.append(f"{entity} {result['relation']} {result['object']}")
return "\n".join(facts)
def respond(self, user_input: str) -> str:
"""Generate response using both memory types"""
# Retrieve relevant episodic memories
past_conversations = self.retrieve_episodic(user_input)
# Extract entities and retrieve semantic facts
entity_prompt = f"Extract the main entity from: {user_input}. Respond with only the entity name."
entity_response = self.llm.invoke(entity_prompt)
entity = entity_response.content.strip()
semantic_facts = self.retrieve_semantic(entity)
# Generate response with context
response_prompt = f"""User: {user_input}
Relevant past conversations:
{chr(10).join(past_conversations)}
Known facts:
{semantic_facts}
Provide a helpful response using this context."""
response = self.llm.invoke(response_prompt)
# Store this interaction
self.add_episodic_memory(f"User: {user_input}\nAssistant: {response.content}")
return response.content
# Usage
agent = DualMemoryAgent()
# Add some semantic facts
agent.add_semantic_fact("Python", "IS_A", "Programming Language")
agent.add_semantic_fact("Python", "CREATED_BY", "Guido van Rossum")
agent.add_semantic_fact("Python", "USED_FOR", "Data Science")
# Interact with memory-enhanced agent
response = agent.respond("Tell me about Python")
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 import TavilySearchResults
llm = ChatOpenAI(model="gpt-4")
search = TavilySearchResults(max_results=5)
# Perform search
search_results = search.run(state['user_input'])
# Synthesize response
synthesis_prompt = f"""User question: {state['user_input']}
Search results:
{search_results}
Provide a comprehensive, well-sourced answer."""
response = llm.invoke(synthesis_prompt)
return {"final_response": response.content}
def coding_specialist(state: MetaControllerState):
"""Handle coding and technical queries"""
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4", temperature=0.2)
coding_prompt = f"""You are an expert programmer. Help with this request:
{state['user_input']}
Provide clean, well-documented code with explanations."""
response = llm.invoke(coding_prompt)
return {"final_response": response.content}
def writing_specialist(state: MetaControllerState):
"""Handle creative and content writing"""
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4", temperature=0.8)
writing_prompt = f"""You are a skilled writer. Help with this request:
{state['user_input']}
Be creative, engaging, and well-structured."""
response = llm.invoke(writing_prompt)
return {"final_response": response.content}
def general_specialist(state: MetaControllerState):
"""Handle general conversation"""
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4")
response = llm.invoke(state['user_input'])
return {"final_response": response.content}
def route_to_specialist(state: MetaControllerState) -> Literal["research", "coding", "writing", "general"]:
"""Route based on classified agent type"""
return state["agent_type"].value
# Build meta-controller workflow
workflow = StateGraph(MetaControllerState)
workflow.add_node("controller", meta_controller)
workflow.add_node("research", research_specialist)
workflow.add_node("coding", coding_specialist)
workflow.add_node("writing", writing_specialist)
workflow.add_node("general", general_specialist)
workflow.set_entry_point("controller")
workflow.add_conditional_edges(
"controller",
route_to_specialist,
{
"research": "research",
"coding": "coding",
"writing": "writing",
"general": "general"
}
)
workflow.add_edge("research", END)
workflow.add_edge("coding", END)
workflow.add_edge("writing", END)
workflow.add_edge("general", END)
app = workflow.compile()
# Use meta-controller
result = app.invoke({
"user_input": "What are the latest developments in quantum computing?",
"agent_type": AgentType.GENERAL,
"final_response": ""
})
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"] < 3:
return "retry"
elif state["error"]:
return "failed"
return "success"
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
- Start with basics:
01_reflection.ipynb,02_tool_use.ipynb,03_ReAct.ipynb - Multi-agent fundamentals:
05_multi_agent.ipynb,11_meta_controller.ipynb - Advanced memory:
08_episodic_with_semantic.ipynb,12_graph.ipynb - Safety patterns:
06_PEV.ipynb,14_dry_run.ipynb,17_reflexive_metacognitive.ipynb - 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()