undefined

undefined

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

undefined

undefined

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"

undefined

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

undefined

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

undefined

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(" -> ")}

undefined

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

undefined

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

undefined

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"

undefined

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

undefined

from langgraph.checkpoint.sqlite import SqliteSaver

undefined

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)

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"

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

jupyter notebook

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

from langchain_openai import ChatOpenAI
from langchain.callbacks import get_openai_callback

undefined

undefined

undefined

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()