LangGraph Integration

pip install graphiti-core langchain-openai langgraph

# For Jupyter notebooks
pip install ipywidgets

# Neo4j via Docker
docker compose up

# Or FalkorDB
docker run -p 6379:6379 -p 3000:3000 -it --rm falkordb/falkordb:latest

import os
from datetime import datetime, timezone
from graphiti_core import Graphiti
from graphiti_core.nodes import EpisodeType

# Configure Graphiti
neo4j_uri = os.getenv('NEO4J_URI', 'bolt://localhost:7687')
neo4j_user = os.getenv('NEO4J_USER', 'neo4j')
neo4j_password = os.getenv('NEO4J_PASSWORD', 'password')

client = Graphiti(neo4j_uri, neo4j_user, neo4j_password)
await client.build_indices_and_constraints()

import json
from pathlib import Path

async def ingest_products_data(client: Graphiti):
    # Load product JSON
    with open('products.json') as f:
        products = json.load(f)['products']
    
    # Add each product as an episode
    for i, product in enumerate(products):
        await client.add_episode(
            name=product.get('title', f'Product {i}'),
            episode_body=str({k: v for k, v in product.items() if k != 'images'}),
            source_description='Product catalog',
            source=EpisodeType.json,
            reference_time=datetime.now(timezone.utc),
        )

await ingest_products_data(client)

from graphiti_core.search.search_config_recipes import NODE_HYBRID_SEARCH_EPISODE_MENTIONS

user_name = 'jess'

# Add initial user episode
await client.add_episode(
    name='User Creation',
    episode_body=f'{user_name} is interested in buying running shoes',
    source=EpisodeType.text,
    reference_time=datetime.now(timezone.utc),
    source_description='SalesBot',
)

# Get user node UUID for centered searches
node_list = await client._search(user_name, NODE_HYBRID_SEARCH_EPISODE_MENTIONS)
user_node_uuid = node_list.nodes[0].uuid

# Get product brand node UUID
node_list = await client._search('ManyBirds', NODE_HYBRID_SEARCH_EPISODE_MENTIONS)
brand_node_uuid = node_list.nodes[0].uuid

from langchain_core.tools import tool
from graphiti_core.edges import EntityEdge

def edges_to_facts_string(entities: list[EntityEdge]):
    """Convert entity edges to readable facts."""
    return '-' + '\n- '.join([edge.fact for edge in entities])

@tool
async def get_shoe_data(query: str) -> str:
    """Search the knowledge graph for information about shoes."""
    edge_results = await client.search(
        query,
        center_node_uuid=brand_node_uuid,  # Center search on product brand
        num_results=10,
    )
    return edges_to_facts_string(edge_results)

tools = [get_shoe_data]

from langchain_core.messages import AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from typing import Annotated
from typing_extensions import TypedDict
import asyncio

class State(TypedDict):
    messages: Annotated[list, add_messages]
    user_name: str
    user_node_uuid: str

async def chatbot(state: State):
    """Agent node that retrieves context from Graphiti and generates responses."""
    facts_string = None
    
    if len(state['messages']) > 0:
        last_message = state['messages'][-1]
        query = f'{"SalesBot" if isinstance(last_message, AIMessage) else state["user_name"]}: {last_message.content}'
        
        # Search Graphiti using user's node as center
        # Facts closer to the user node are ranked higher
        edge_results = await client.search(
            query,
            center_node_uuid=state['user_node_uuid'],
            num_results=5
        )
        facts_string = edges_to_facts_string(edge_results)
    
    # Build system message with retrieved facts
    system_message = SystemMessage(
        content=f"""You are a skillful shoe salesperson working for ManyBirds.
        Review information about the user and their conversation below.
        Keep responses short and concise. Always be selling and helpful!
        
        Things you need to know to close a sale:
        - User's shoe size
        - Any special needs (wide feet, arch support, etc.)
        - Preferred colors and styles
        - Budget
        
        Facts about the user:
        {facts_string or 'No facts about the user yet'}"""
    )
    
    messages = [system_message] + state['messages']
    response = await llm.ainvoke(messages)
    
    # Asynchronously persist interaction to Graphiti
    asyncio.create_task(
        client.add_episode(
            name='Chatbot Response',
            episode_body=f'{state["user_name"]}: {state["messages"][-1].content}\nSalesBot: {response.content}',
            source=EpisodeType.message,
            reference_time=datetime.now(timezone.utc),
            source_description='Chatbot',
        )
    )
    
    return {'messages': [response]}

# Initialize LLM with tools
llm = ChatOpenAI(model='gpt-4.1-mini', temperature=0).bind_tools(tools)
tool_node = ToolNode(tools)

# Define conditional logic
async def should_continue(state, config):
    last_message = state['messages'][-1]
    return 'end' if not last_message.tool_calls else 'continue'

# Build the graph
graph_builder = StateGraph(State)
memory = MemorySaver()

graph_builder.add_node('agent', chatbot)
graph_builder.add_node('tools', tool_node)

graph_builder.add_edge(START, 'agent')
graph_builder.add_conditional_edges(
    'agent',
    should_continue,
    {'continue': 'tools', 'end': END}
)
graph_builder.add_edge('tools', 'agent')

graph = graph_builder.compile(checkpointer=memory)

import uuid

# Single interaction
response = await graph.ainvoke(
    {
        'messages': [{
            'role': 'user',
            'content': 'What sizes do the Wool Runners come in?'
        }],
        'user_name': user_name,
        'user_node_uuid': user_node_uuid,
    },
    config={'configurable': {'thread_id': uuid.uuid4().hex}}
)

print(response['messages'][-1].content)

config = {'configurable': {'thread_id': uuid.uuid4().hex}}
user_state = {'user_name': user_name, 'user_node_uuid': user_node_uuid}

print("Agent: Hello! How can I help you find shoes today?")

while True:
    user_input = input("You: ")
    if user_input.lower() in ['exit', 'quit', 'bye']:
        print("Agent: Thank you for chatting! Goodbye!")
        break
    
    graph_state = {
        'messages': [{'role': 'user', 'content': user_input}],
        'user_name': user_state['user_name'],
        'user_node_uuid': user_state['user_node_uuid'],
    }
    
    async for event in graph.astream(graph_state, config=config):
        for value in event.values():
            if 'messages' in value:
                last_message = value['messages'][-1]
                if isinstance(last_message, AIMessage) and isinstance(last_message.content, str):
                    print(f"Agent: {last_message.content}")

# Center search on user node for personalized context
user_context = await client.search(
    query=user_message,
    center_node_uuid=user_node_uuid,
    num_results=5
)

# Center search on domain entities for factual information
product_info = await client.search(
    query=product_query,
    center_node_uuid=product_category_uuid,
    num_results=10
)

import asyncio

# Don't await - fire and forget
asyncio.create_task(
    client.add_episode(
        name='Interaction',
        episode_body=conversation_text,
        source=EpisodeType.message,
        reference_time=datetime.now(timezone.utc),
    )
)

# LangGraph manages immediate conversation flow
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# Graphiti stores extractable knowledge
await client.add_episode(...)  # Persisted facts

# Sales agent adds customer preference
await client.add_episode(
    name='Customer Preference',
    episode_body='Customer prefers blue shoes',
    group_id='customer-123',
    source=EpisodeType.text,
)

# Support agent retrieves the preference later
context = await client.search(
    'customer preferences',
    group_id='customer-123',
    num_results=5
)

from langchain_core.tools import tool

def create_category_search_tool(category_name: str, category_uuid: str):
    @tool
    async def search_category(query: str) -> str:
        f"""Search for {category_name} information."""
        results = await client.search(
            query,
            center_node_uuid=category_uuid,
            num_results=10
        )
        return edges_to_facts_string(results)
    
    return search_category

# Create tools for each category
categories = [
    ('Running Shoes', 'uuid-running'),
    ('Hiking Boots', 'uuid-hiking'),
    ('Casual Sneakers', 'uuid-casual'),
]

tools = [create_category_search_tool(name, uuid) for name, uuid in categories]

from datetime import timedelta

# Search for recent interactions
recent_time = datetime.now(timezone.utc) - timedelta(days=7)
recent_context = await client.search(
    query='customer interactions',
    reference_time=recent_time,
    num_results=10
)

# System message with temporal context
system_message = SystemMessage(
    content=f"""Recent customer activity (last 7 days):
    {edges_to_facts_string(recent_context)}
    
    Acknowledge any recent purchases or concerns."""
)

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

class CustomerProfile(BaseModel):
    shoe_size: int | None = Field(description="Customer's shoe size")
    preferred_colors: list[str] = Field(description="Preferred colors")
    budget_range: str | None = Field(description="Budget range")
    special_needs: list[str] = Field(description="Special requirements")

parser = PydanticOutputParser(pydantic_object=CustomerProfile)

# Extract structured data from graph context
edges = await client.search(f'customer profile for {user_name}', num_results=10)
facts = edges_to_facts_string(edges)

prompt = f"""Extract customer profile from these facts:
{facts}

{parser.get_format_instructions()}
"""

response = await llm.ainvoke(prompt)
profile = parser.parse(response.content)

// View user's preferences in Neo4j Browser
MATCH (u:Entity {name: 'jess'})-[r:RELATES_TO]->(n)
RETURN u, r, n
LIMIT 25

(jess:Entity)-[:INTERESTED_IN]->(Running Shoes:Entity)
(jess)-[:PREFERS]->(Size 10:Entity)
(jess)-[:LIKES]->(Blue Color:Entity)
(Running Shoes)-[:HAS_FEATURE]->(Lightweight:Entity)
(Running Shoes)-[:MANUFACTURED_BY]->(ManyBirds:Entity)

# Personalized search centered on user
user_facts = await client.search(
    query,
    center_node_uuid=user_node_uuid,
    num_results=5
)

# Domain search centered on topic
domain_facts = await client.search(
    query,
    center_node_uuid=domain_node_uuid,
    num_results=10
)

# Separate graphs per customer
await client.add_episode(
    name='Interaction',
    episode_body=content,
    group_id=f'customer-{customer_id}',
    source=EpisodeType.text,
)

# Search within customer's graph
results = await client.search(
    query,
    group_id=f'customer-{customer_id}',
    num_results=10
)

# For development/low-tier APIs
export SEMAPHORE_LIMIT=5

# For production/high-tier APIs
export SEMAPHORE_LIMIT=20

async def safe_add_episode(content: str):
    try:
        await client.add_episode(
            name='Interaction',
            episode_body=content,
            source=EpisodeType.text,
            reference_time=datetime.now(timezone.utc),
        )
    except Exception as e:
        logger.error(f"Failed to persist episode: {e}")
        # Continue agent operation even if persistence fails

class CustomerServiceState(TypedDict):
    messages: Annotated[list, add_messages]
    customer_id: str
    customer_node_uuid: str
    current_agent: str  # 'sales', 'support', 'billing'

async def route_to_agent(state: CustomerServiceState):
    """Route to appropriate agent based on intent."""
    last_message = state['messages'][-1].content
    
    # Retrieve customer context
    context = await client.search(
        f"customer {state['customer_id']} {last_message}",
        center_node_uuid=state['customer_node_uuid'],
        num_results=10
    )
    
    # Determine routing based on context
    intent_prompt = f"""Given this customer context:
    {edges_to_facts_string(context)}
    
    And their message: {last_message}
    
    Route to: sales, support, or billing
    """
    
    routing = await llm.ainvoke(intent_prompt)
    return routing.content.lower()

# Build multi-agent graph
graph_builder = StateGraph(CustomerServiceState)
graph_builder.add_node('router', route_to_agent)
graph_builder.add_node('sales', sales_agent)
graph_builder.add_node('support', support_agent)
graph_builder.add_node('billing', billing_agent)

graph_builder.add_edge(START, 'router')
graph_builder.add_conditional_edges(
    'router',
    lambda s: s['current_agent'],
    {
        'sales': 'sales',
        'support': 'support',
        'billing': 'billing'
    }
)

import os

# Enable tracing
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_PROJECT'] = 'Graphiti-LangGraph-Agent'
os.environ['LANGCHAIN_API_KEY'] = 'your-langsmith-key'

# Run agent - traces automatically captured
await graph.ainvoke(...)