# From RAG to Agentic AI. How I Added LangGraph to My Local

> Source: <https://dev.to/josaphatstar/from-rag-to-agentic-ai-how-i-added-langgraph-to-my-local-a6e>
> Published: 2026-07-29 21:42:11+00:00

In my [previous article](https://dev.to/josaphatstar/i-built-a-local-rag-assistant-with-ollama-chromadb-and-langchain-heres-what-i-learned-5a2e), I built a fully local RAG assistant Ollama, ChromaDB, LangChain, all running in Docker. It answered technical support questions by searching through documentation and citing sources. It worked.

But after using it for a while, I noticed something uncomfortable: **it treated every question the same way**.

Ask it *"how to close monthly payroll?"* it searches the docs. Fine.

Ask it *"the server crashes at startup"* it also searches the docs. Less fine.

Ask it something completely outside the documentation it searches the docs. Useless.

A real support technician doesn't do that. They first *assess* the situation, then decide what to do: look it up, run a diagnosis, or escalate to a human. My RAG had no such judgment.

That's what this article is about how I evolved the system into an **Agentic AI architecture** using LangGraph, where the assistant first decides *which strategy to use*, then acts accordingly.

Classic RAG is a linear pipeline. Every query follows the exact same path:

```
Question → Embed → Retrieve → Prompt → LLM → Answer
```

No branching. No decision-making. No memory between steps.

This works perfectly for procedural questions where the answer lives in the docs. But technical support involves at least three distinct scenarios:

| Scenario | Example | Best strategy |
|---|---|---|
| Procedural question | "How do I create an account?" | Search documentation |
| Known error code | "ERR-COMP-001 appears" | Lookup error database |
| Unknown incident | "Server crashes, no idea why" | Diagnose + escalate if needed |

A single RAG pipeline handles the first case well and the other two poorly. The solution is to add a layer of reasoning *before* retrieval.

The shift from RAG to Agentic AI comes down to one thing: **the system plans before it acts**.

Instead of one fixed pipeline, you have:

```
Question
    ↓
Classifier (what kind of question is this?)
    ↓
    ├── Procedural → RAG Agent (search docs)
    ├── Error code → Diagnostic Agent (lookup + LLM analysis)
    └── Complex    → Diagnostic Agent → Escalation Agent (generate ticket)
```

Each branch is an **agent**; an autonomous function that receives a shared state, does its job, updates the state, and returns it. LangGraph connects these agents into a directed graph with conditional routing between them.

Before building agents, you define the state they all share. Think of it as a case file that every agent reads and updates:

``` python
# src/agents/state.py

from typing import TypedDict, List, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    question: str          # Original question never changes
    plan: List[str]
    current_step: int
    past_steps: List[str]  # Audit trail of what each agent did
    response: str
    next_agent: str
    niveau_support: int    # 1=resolved, 2=partial, 3=needs human
```

The `niveau_support`

field is key it's how agents communicate urgency to each other. If an agent sets it to 3, the orchestrator escalates automatically.

`Annotated[Sequence[BaseMessage], operator.add]`

means messages accumulate — each agent appends rather than overwrites. You get a full conversation trace at the end.

The pipeline from my previous article, repackaged as an autonomous agent:

``` php
# src/agents/rag_agent.py

def rag_agent(state: AgentState) -> AgentState:
    """Searches the documentation and generates a grounded answer."""

    docs = vectorstore.similarity_search(state["question"], k=3)

    if not docs:
        return {
            **state,
            "messages": list(state["messages"]) + [
                AIMessage(content="[RAG] No relevant document found.")
            ],
            "past_steps": state["past_steps"] + ["RAG: no results"],
            "niveau_support": 3   # Nothing found — escalate
        }

    context = "\n\n".join([
        f"[Source: {d.metadata.get('module', '?')}]\n{d.page_content}"
        for d in docs
    ])

    answer = llm.invoke(
        f"Answer from context only.\nContext: {context}\nQuestion: {state['question']}"
    )

    return {
        **state,
        "messages": list(state["messages"]) + [AIMessage(content=f"[RAG] {answer}")],
        "past_steps": state["past_steps"] + ["RAG: documentation search completed"],
        "niveau_support": 1
    }
```

The critical difference from the classic RAG: the agent returns an **updated state**, not just an answer. Every agent follows this same pattern — receive state, do work, return updated state.

Handles error codes and technical incidents with two modes:

```
# src/agents/diagnostic_agent.py

KNOWN_ERRORS = {
    "ERR-COMP-001": {
        "description": "Debit/credit imbalance",
        "solution": "Check Total Debit = Total Credit. Go to Accounting > Journal Entry."
    },
    "ERR-PAIE-042": {
        "description": "Payroll line missing rate",
        "solution": "Go to Settings > Payroll Lines, fill in the missing rate."
    },
}

def diagnostic_agent(state: AgentState) -> AgentState:
    question = state["question"]
    found_codes = [code for code in KNOWN_ERRORS if code.lower() in question.lower()]

    if found_codes:
        # Mode 1 : known error: instant answer, no LLM call needed
        responses = [
            f"Code {code}: {KNOWN_ERRORS[code]['description']}\n"
            f"Solution: {KNOWN_ERRORS[code]['solution']}"
            for code in found_codes
        ]
        return {
            **state,
            "messages": list(state["messages"]) + [
                AIMessage(content=f"[Diagnostic] {chr(10).join(responses)}")
            ],
            "past_steps": state["past_steps"] + ["Diagnostic: known error, resolved"],
            "niveau_support": 1
        }

    # Mode 2 : unknown error: LLM analysis
    answer = llm.invoke(
        f"Analyze this technical issue. If unsure, say so.\nIssue: {question}"
    )
    return {
        **state,
        "messages": list(state["messages"]) + [AIMessage(content=f"[Diagnostic] {answer}")],
        "past_steps": state["past_steps"] + ["Diagnostic: LLM analysis, uncertain"],
        "niveau_support": 2
    }
```

The known error lookup runs in milliseconds without touching the LLM; a meaningful performance gain for common incidents.

Called only when `niveau_support >= 3`

. It doesn't try to answer; it generates a structured ticket with the full diagnostic history already attached:

``` php
# src/agents/escalade_agent.py

def escalade_agent(state: AgentState) -> AgentState:
    ticket = f"""
=== SUPPORT TICKET — LEVEL 3 ===
Date     : {datetime.now().strftime("%Y-%m-%d %H:%M")}
Priority : HIGH

ISSUE:
{state["question"]}

DIAGNOSTIC STEPS ALREADY TAKEN:
{chr(10).join(state["past_steps"])}

CONCLUSION: Human intervention required.
================================"""

    return {
        **state,
        "messages": list(state["messages"]) + [AIMessage(content=ticket)],
        "past_steps": state["past_steps"] + ["Escalation: level 3 ticket generated"],
        "niveau_support": 3
    }
```

This is what separates a real support system from a chatbot; it knows when to stop and hand off, with context already attached so the user doesn't have to repeat themselves.

``` php
# src/agents/orchestrator.py

def classifier(state: AgentState) -> str:
    question = state["question"].lower()

    if re.search(r'err-\w+-\d+', question):
        return "diagnostic"

    problem_words = ["error", "crash", "bug", "not working", "fails", "problem"]
    if any(word in question for word in problem_words):
        return "diagnostic"

    return "rag"  # Default for procedural questions

def should_escalate(state: AgentState) -> str:
    return "escalade" if state.get("niveau_support", 1) >= 3 else END

def create_orchestrator():
    workflow = StateGraph(AgentState)

    workflow.add_node("rag_agent", rag_agent)
    workflow.add_node("diagnostic_agent", diagnostic_agent)
    workflow.add_node("escalade_agent", escalade_agent)

    workflow.set_conditional_entry_point(
        classifier,
        {"rag": "rag_agent", "diagnostic": "diagnostic_agent"}
    )

    workflow.add_edge("rag_agent", END)
    workflow.add_conditional_edges(
        "diagnostic_agent",
        should_escalate,
        {"escalade": "escalade_agent", END: END}
    )
    workflow.add_edge("escalade_agent", END)

    return workflow.compile()
```

Three questions, three different paths:

**Procedural question**

```
Input  : "How to close monthly payroll?"
Path   : classifier → rag_agent → END
Output : agents_used: ["RAG: documentation search completed"]
         niveau: 1
```

**Known error code**

```
Input  : "ERR-COMP-001 appears on journal validation"
Path   : classifier → diagnostic_agent → END
Output : agents_used: ["Diagnostic: known error, resolved"]
         niveau: 1  (no LLM call, instant response)
```

**Unknown incident**

```
Input  : "Server crashes at startup, no idea why"
Path   : classifier → diagnostic_agent → escalade_agent → END
Output : agents_used: ["Diagnostic: LLM analysis", "Escalation: ticket generated"]
         niveau: 3
```

**The classifier is fragile.** Keyword matching breaks on edge cases *"I don't have a problem, just need help configuring payroll"* contains "problem" and routes to diagnostic instead of RAG. A small LLM classification call, or a fine-tuned classifier, would be more robust.

**Agents reload the embedding model on every call.** Instantiating `HuggingFaceEmbeddings`

on each agent invocation adds unnecessary overhead. Initialize once at startup and pass the instance through the state.

**The escalation threshold is hardcoded.** A feedback loop where user ratings influence the confidence threshold over time would make this genuinely adaptive that's what reinforcement learning adds in the next phase of this project.

| Classic RAG | Agentic AI | |
|---|---|---|
Best for |
Uniform question types | Mixed question types |
Complexity |
Low | Medium-High |
Latency |
One LLM call | One or more LLM calls |
Flexibility |
Fixed pipeline | Extensible — add agents without rewriting |
When to choose |
Start here | When RAG fails on edge cases |

My honest take: **start with RAG**. Get it working, understand its failure modes, then add agents where you actually need branching logic. Adding LangGraph too early is over-engineering.

**Design your AgentState first.** It's the contract between all components. Changing it mid-project means updating every agent get it right upfront.

**Agents should do one thing.** The moment an agent tries to retrieve *and* diagnose, it becomes untestable and unroutable. Keep them single-responsibility.

** past_steps is your audit trail.** When a ticket escalates to a human, they see exactly what the system tried. Small design choice, professional result.

**The classifier is your most important and most brittle piece of code.** Test it with as many edge cases as you can before trusting it.

These two articles cover Phases 1 and 2 of my internship project. The logical next step is a reinforcement learning feedback loop; user ratings adjusting the classifier's routing decisions over time; which I may write about once it's implemented and tested.

I'm a Master's student in AI & Big Data based, building these systems with open-source tools and no GPU. If you're working on multi-agent systems or ran into similar design decisions; especially around the classifier or state design; I'd love to hear your approach in the comments.