all-agentic-architectures

# All Agentic Architectures Skill > Skill by [ara.so](https://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 ```bash # 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 ```bash 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: ```bash # 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. ```python 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. ```python 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. ```python 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. ```python 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": [],