LangGraph

pip install langgraph langchain-openai supermemory python-dotenv

# .env
SUPERMEMORY_API_KEY=your-supermemory-api-key
OPENAI_API_KEY=your-openai-api-key

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from supermemory import Supermemory
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o")
memory = Supermemory()

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

def agent(state: State):
    user_id = state["user_id"]
    messages = state["messages"]
    user_query = messages[-1].content

    # Fetch user profile with relevant memories
    profile_result = memory.profile(container_tag=user_id, q=user_query)

    # Build context from profile
    static_facts = profile_result.profile.static or []
    dynamic_context = profile_result.profile.dynamic or []
    search_results = profile_result.search_results.results if profile_result.search_results else []

    context = f"""
User Background:
{chr(10).join(static_facts) if static_facts else 'No profile yet.'}

Recent Context:
{chr(10).join(dynamic_context) if dynamic_context else 'No recent activity.'}

Relevant Memories:
{chr(10).join([r.memory or r.chunk for r in search_results]) if search_results else 'None found.'}
"""

    system = SystemMessage(content=f"You are a helpful assistant.\n\n{context}")
    response = llm.invoke([system] + messages)

    # Store the interaction
    memory.add(
        content=f"User: {user_query}\nAssistant: {response.content}",
        container_tag=user_id
    )

    return {"messages": [response]}

# Build the graph
graph = StateGraph(State)
graph.add_node("agent", agent)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)
app = graph.compile()

# Run it
result = app.invoke({
    "messages": [HumanMessage(content="Hi! I'm working on a Python project.")],
    "user_id": "user_123"
})
print(result["messages"][-1].content)

result = memory.profile(
    container_tag="user_123",
    q="optional search query"  # Also returns relevant memories
)

print(result.profile.static)   # ["User is a Python developer", "Prefers functional style"]
print(result.profile.dynamic)  # ["Working on async patterns", "Debugging rate limiting"]

# Store a conversation
memory.add(
    content="User asked about graph traversal. Explained BFS vs DFS.",
    container_tag="user_123",
    metadata={"topic": "algorithms", "type": "conversation"}
)

# Store a document
memory.add(
    content="https://langchain-ai.github.io/langgraph/",
    container_tag="user_123"
)

results = memory.search.memories(
    q="graph algorithms",
    container_tag="user_123",
    search_mode="hybrid",
    limit=5
)

for r in results.results:
    print(r.memory or r.chunk, r.similarity)

from typing import Annotated, TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from supermemory import Supermemory
from dotenv import load_dotenv

load_dotenv()

class SupportAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
        self.memory = Supermemory()
        self.app = self._build_graph()

    def _build_graph(self):
        class State(TypedDict):
            messages: Annotated[list, add_messages]
            user_id: str
            context: str
            category: Optional[str]

        def retrieve_context(state: State):
            """Fetch user profile and relevant past tickets."""
            user_id = state["user_id"]
            query = state["messages"][-1].content

            result = self.memory.profile(
                container_tag=user_id,
                q=query,
                threshold=0.5
            )

            static = result.profile.static or []
            dynamic = result.profile.dynamic or []
            memories = result.search_results.results if result.search_results else []

            context = f"""
## User Profile
{chr(10).join(f"- {fact}" for fact in static) if static else "New user, no history."}

## Current Context
{chr(10).join(f"- {ctx}" for ctx in dynamic) if dynamic else "No recent activity."}

## Related Past Tickets
{chr(10).join(f"- {m.memory}" for m in memories[:3]) if memories else "No similar issues found."}
"""
            return {"context": context}

        def categorize(state: State):
            """Determine ticket category for routing."""
            query = state["messages"][-1].content.lower()

            if any(word in query for word in ["billing", "payment", "charge", "invoice"]):
                return {"category": "billing"}
            elif any(word in query for word in ["bug", "error", "broken", "crash"]):