Building a LangGraph RAG Agent from Scratch — with a Live UI That Shows Every Step

frontend/   ← React + Vite chat UI (live agent loop visualisation)
backend/    ← FastAPI server wrapping the RAG agent
step*.py    ← 6 progressive learning files

from langchain_groq import ChatGroq
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatGroq(model="llama-3.3-70b-versatile")

messages = [
    SystemMessage(content="You are a rate limiting expert."),
    HumanMessage(content="What is token bucket?"),
]

response = llm.invoke(messages)
print(response.content)

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a rate limiting expert."),
    ("human", "{question}"),
])

# Chain: prompt → LLM
chain = prompt | llm

# Invoke
response = chain.invoke({"question": "Compare token bucket and leaky bucket"})

# Stream tokens as they arrive
for chunk in chain.stream({"question": "What is sliding window log?"}):
    print(chunk.content, end="", flush=True)

from langchain_core.tools import tool
from langchain_groq import ChatGroq

@tool
def get_algorithm_info(algorithm: str) -> str:
    """Return a brief description of a rate limiting algorithm."""
    descriptions = {
        "token_bucket":    "Tokens refill at a fixed rate up to a capacity cap. Allows bursts.",
        "fixed_window":    "Counts requests in fixed time windows. Simple but has boundary spikes.",
        "sliding_window":  "Precise per-request log. High memory, no boundary spikes.",
        "leaky_bucket":    "Queue drains at a constant rate. Smooths traffic, no bursts allowed.",
    }
    return descriptions.get(algorithm, "Unknown algorithm.")

# Bind tools to the model — it now knows what tools exist and their signatures
llm_with_tools = ChatGroq(model="meta-llama/llama-4-scout-17b-16e-instruct").bind_tools(
    [get_algorithm_info]
)

response = llm_with_tools.invoke("Tell me about token bucket")
# response.tool_calls → [{"name": "get_algorithm_info", "args": {"algorithm": "token_bucket"}}]

from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    messages: Annotated[list, add_messages]  # reducer: appends, never replaces

def node_a(state: State):
    return {"messages": ["Hello from node A"]}

def node_b(state: State):
    return {"messages": ["Hello from node B"]}

def route(state: State):
    return "b" if len(state["messages"]) < 3 else END

graph = StateGraph(State)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.set_entry_point("a")
graph.add_conditional_edges("a", route, {"b": "b", END: END})
graph.add_edge("b", "a")

app = graph.compile()

from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage
from typing import Annotated
from typing_extensions import TypedDict

tools = [get_algorithm_info, recommend_algorithm, calculate_token_bucket]
llm   = ChatGroq(model="meta-llama/llama-4-scout-17b-16e-instruct").bind_tools(tools)

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

def llm_node(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

def tools_condition(state: State):
    return "tools" if state["messages"][-1].tool_calls else END

graph = StateGraph(State)
graph.add_node("llm",   llm_node)
graph.add_node("tools", ToolNode(tools))
graph.set_entry_point("llm")
graph.add_conditional_edges("llm", tools_condition)
graph.add_edge("tools", "llm")  # always loop back after tool execution

agent = graph.compile()

result = agent.invoke({"messages": [HumanMessage(content="What algorithm for bursty traffic?")]})
print(result["messages"][-1].content)

START → [llm] → has tool_calls? → YES → [tools] → back to [llm]
                                → NO  → END

from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.tools import tool

# Index documents once at startup
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

docs = load_knowledge_base()           # returns list of Document objects
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(docs)

vectorstore = FAISS.from_documents(chunks, embeddings)
retriever   = vectorstore.as_retriever(search_kwargs={"k": 3})

# Expose retrieval as a tool
@tool
def search_knowledge_base(query: str) -> str:
    """Search the rate limiting knowledge base for relevant information."""
    docs = retriever.invoke(query)
    return "\n---\n".join(d.page_content for d in docs)

from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage
import json

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        llm_call_count = 0
        graph_started  = False

        async for event in agent.astream_events(
            {"messages": [HumanMessage(content=request.message)]},
            version="v2",
        ):
            kind = event["event"]
            node = event.get("metadata", {}).get("langgraph_node", "")

            # LLM node starting
            if kind == "on_chat_model_start" and node == "llm":
                if not graph_started:
                    graph_started = True
                    yield sse({"type": "pipeline", "phase": "graph_start"})
                llm_call_count += 1
                yield sse({"type": "pipeline", "phase": "llm_start", "call": llm_call_count})

            # LLM done — emit routing decision
            elif kind == "on_chat_model_end" and node == "llm":
                output     = event["data"].get("output")
                tool_calls = getattr(output, "tool_calls", []) if output else []
                yield sse({
                    "type":       "pipeline",
                    "phase":      "llm_end",
                    "decision":   "tools" if tool_calls else "answer",
                    "tool_names": [tc["name"] for tc in tool_calls],
                })

            # Tool executing
            elif kind == "on_tool_start":
                yield sse({"type": "pipeline", "phase": "tool_start",
                           "tool": event["name"], "args": event["data"].get("input", {})})

            # Tool done
            elif kind == "on_tool_end":
                out     = event["data"].get("output", "")
                content = out.content if hasattr(out, "content") else str(out)
                yield sse({"type": "pipeline", "phase": "tool_end",
                           "tool": event["name"], "preview": content[:120]})

            # Individual LLM output tokens (final answer only)
            elif kind == "on_chat_model_stream" and node == "llm":
                chunk = event["data"]["chunk"]
                if chunk.content and not getattr(chunk, "tool_call_chunks", []):
                    yield sse({"type": "token", "content": chunk.content})

        yield sse({"type": "pipeline", "phase": "graph_end"})
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

🚀 StateGraph Initialized          [langgraph]
   StateGraph.compile() · add_messages reducer
   ↓
🧠 LLM Node — Call #1  ⟳           [langchain]   ← spinning while active
   ChatGroq(llama-4-scout) · bind_tools(4)
   AIMessage has tool_calls → selected: search_knowledge_base
   ↓
◆  Conditional Edge → tools node   [langgraph]
   add_conditional_edges · tools_condition(state)
   has tool_calls → route to tools
   ↓
🔍 ToolNode: search_knowledge_base ⟳ [langchain]
   FAISS vector search · HuggingFace embeddings
   query: HTTP headers rate limiting
   → Retrieved 3 relevant chunk(s)
   ↓
🧠 LLM Node — Call #2  ✓           [langchain]
   LLM sees ToolMessage in state
   no tool_calls → generating final answer
   ↓
◆  Conditional Edge → END          [langgraph]
   no tool_calls → route to END
   ↓
🏁 Graph END                       [langgraph]
   messages[-1].content → response

const CHAR_DELAY = 18  // ms per character

// When a token event arrives, push each character into the queue
if (ev.type === 'token') {
  tokenQueue.current.push(...ev.content.split(''))
  startTicker(assistantId)
}

// Ticker drains one char at a time
const startTicker = (id) => {
  tickerRef.current = setInterval(() => {
    if (!tokenQueue.current.length) return
    const ch = tokenQueue.current.shift()
    setMessages(prev => prev.map(m =>
      m.id === id ? { ...m, content: (m.content || '') + ch } : m
    ))
  }, CHAR_DELAY)
}

# Python deps (uses uv to avoid system Python issues)
uv venv .venv --python 3.12
uv pip install -r requirements.txt

# Frontend deps
cd frontend && npm install && cd ..

# Terminal 1 — backend
cd backend
GROQ_API_KEY=your_key uvicorn main:app --port 8000 --reload

# Terminal 2 — frontend
cd frontend && npm run dev