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

查看原文 / View Original 编辑 / Edit

From RAG to Agentic AI: How I Added LangGraph to My Local

在上一篇文章中，我构建了一个完全本地化的 RAG 助手，使用了 Ollama、ChromaDB 和 LangChain，并全部运行在 Docker 中。它通过搜索文档并引用来源来回答技术支持问题。它确实能工作，但在使用一段时间后，我注意到一个令人不适的问题：它对待所有问题的方式都是一样的。问它“如何关闭月度工资单？”它去搜索文档。没问题。问它“服务器启动时崩溃了”，它也去搜索文档。这就不太好了。问它一个完全不在文档范围内的问题，它还是去搜索文档。这毫无用处。真正的技术支持人员不会这样做。他们会先评估情况，然后决定采取什么行动：查阅资料、运行诊断，或者升级给人工处理。我的 RAG 系统缺乏这种判断力。本文的主题就是我如何利用 LangGraph 将该系统演进为 Agentic AI（代理式 AI）架构，让助手先决定使用哪种策略，然后再采取相应的行动。

The Core Limitation of Classic RAG

经典 RAG 的核心局限性

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:

经典 RAG 是一个线性流水线。每个查询都遵循完全相同的路径：问题 → 向量化 → 检索 → 提示词 → 大模型 → 回答。没有分支，没有决策，步骤之间也没有记忆。这对于答案存在于文档中的程序性问题非常有效。但技术支持至少涉及三种不同的场景：

ScenarioExampleBest 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
场景示例最佳策略
程序性问题“我该如何创建账户？”搜索文档
已知错误代码“出现 ERR-COMP-001”查询错误数据库
未知故障“服务器崩溃，原因不明”诊断 + 必要时升级

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.

单一的 RAG 流水线能很好地处理第一种情况，但后两种处理得很差。解决方案是在检索之前增加一层推理。

What Agentic AI Adds

Agentic AI 带来了什么

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:

从 RAG 到 Agentic AI 的转变归结为一件事：系统在行动前会进行规划。你不再拥有一个固定的流水线，而是拥有：

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)

问题 ↓ 分类器（这是什么类型的问题？） ↓ ├── 程序性 → RAG 代理（搜索文档） ├── 错误代码 → 诊断代理（查询 + 大模型分析） └── 复杂问题 → 诊断代理 → 升级代理（生成工单）

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.

每个分支都是一个代理；这是一个自主函数，它接收共享状态，完成工作，更新状态并将其返回。LangGraph 将这些代理连接成一个有向图，并在它们之间进行条件路由。

The Shared State Foundation of LangGraph

LangGraph 的共享状态基础

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

在构建代理之前，你需要定义它们共享的状态。可以把它想象成每个代理都会读取和更新的“案卷”：

# 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.

niveau_support 字段是关键，它是代理之间沟通紧急程度的方式。如果某个代理将其设置为 3，编排器会自动进行升级。Annotated[Sequence[BaseMessage], operator.add] 意味着消息会累积——每个代理都是追加而不是覆盖。最终你会得到完整的对话轨迹。

Agent 1: The RAG Agent

代理 1：RAG 代理

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

将我上一篇文章中的流水线重新封装为一个自主代理：

# 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.

与经典 RAG 的关键区别在于：代理返回的是更新后的状态，而不仅仅是一个答案。每个代理都遵循相同的模式——接收状态、执行工作、返回更新后的状态。

Agent 2: The Diagnostic Agent

代理 2：诊断代理

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..."},
    "ERR-PAIE-042": {"description": "Payroll line missing rate", "solution": "Go to Settings > Payroll Lines..."},
}

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']}\nSolution: {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.

已知错误查询在几毫秒内即可完成，无需调用大模型；这对于常见故障来说是一个显著的性能提升。

Agent 3: The Escalation Agent

代理 3：升级代理

Called only when niveau_support >= 3. It doesn’t try to answer; it generates a structured ticket with the full diagnostic history already attached:

仅在 niveau_support >= 3 时调用。它不会尝试回答，而是生成一个结构化的工单，并附带完整的诊断历史记录：

# src/agents/escalation_agent.py
# ... (Implementation details)
← 返回新闻列表 / Back to news list