Home / Articles / Search Runbooks, Don't Paste Them: A Small Cited RAG Path in LangChain

This article is published in English.

langchain rag retrieval vector-search python

Search Runbooks, Don't Paste Them: A Small Cited RAG Path in LangChain

Build a two-step retrieval path in LangChain: load runbooks, chunk them, search with a focused query, keep metadata, require cited source IDs and test it without paid embeddings.

2886 words

Search the runbooks instead of stuffing the prompt

By this point the helpdesk app branches by ticket type. A database ticket pulls in the service catalog, recent incidents and recent deployments at once, and an access ticket goes down another route. Still missing are the detailed how-to steps behind a recommendation.

You could paste every runbook into the prompt. For three short files that works, but it stops working as the collection grows. Big prompts cost more, run slower, and make it hard to tell which passage backed an answer. They can also hand the model documents the current user was never allowed to read. The answer is search.

The path built here has two phases. At indexing time, Markdown runbooks become Document objects carrying source metadata, and those split into smaller chunks:

indexing time
Markdown runbooks
  -> Document objects with source metadata
  -> smaller chunks
  -> embeddings
  -> vector store

At request time, the ticket classification produces a focused search query, which goes to the retriever:

request time
ticket classification
  -> focused search query
  -> retriever
  -> a few runbook passages with source IDs
  -> recommendation chain
  -> source-ID check

People often call this two-step retrieval-augmented generation, or two-step RAG. The label matters less than the boundary: search comes first, and generation only receives the passages that came back. The model never decides whether a search happens, and it can't pick an arbitrary data source. No agent is needed.

The ticket used throughout

The running example is a ticket reading: EU checkout requests fail after the 14:05 deployment, and the checkout-api logs show "database connection refused." Its structured classification looks like this:

{
  "ticket_id": "INC-2048",
  "summary": "EU checkout fails after a deployment.",
  "category": "database",
  "severity": "SEV2",
  "service": "checkout-api",
  "confidence": 0.91,
  "evidence": ["database connection refused"],
  "missing_information": ["Affected request percentage"]
}

A good search returns the checkout database runbook, and not a payment-latency guide or an identity-access guide. It should also keep enough metadata to answer four questions: which file the text came from, which service owns it, where in the file the chunk started, and which source IDs the recommendation cited. Text with no source metadata is hard to operate.

The six parts of retrieval

Each part needs a definition before it's used.

Document. Think of a Document as a block of text with labels attached. The text can originate in a Markdown file, a PDF, a web page, a wiki or even a row in a table, and the labels can record the source ID, title, service, path, owner, version or who may see it.

Chunk. Documents get cut into pieces that can be found independently. A single runbook may describe diagnosis, recovery and rollback together, yet a question about database credentials has no reason to return the rollback material each time.

Embedding. An embedding model turns text into a list of numbers, and texts with similar meaning should end up near each other under the chosen comparison method. It isn't a fact check. Similar text can still be wrong, outdated or off limits.

Vector store. A vector store holds document vectors and finds nearby ones. Some live only in memory, others sit on a database or a hosted service. Where you store vectors doesn't replace access control.

Retriever. Hand a retriever free-form query text and it gives back Document objects. It's a wider idea than vector search: behind the same interface you could put keyword matching, a vector store, a hybrid search service or an in-house search API.

Generation. The final chain gets the chosen passages and produces a typed answer. Whatever it writes, it can't pretend the search returned something it didn't.

Retrieval is not memory

A set of runbooks is a body of knowledge, and querying it counts as retrieval. Chat history is a sequence of messages, saved user facts are ordinary application data, and workflow checkpoints record execution state. Their retention and permission rules all differ, which is why the next stage pulls them apart in code.

Project layout and setup

The snapshot for this stage has this structure:

langchain-helpdesk/
├── app.py
├── chains.py
├── embeddings.py
├── retrieval.py
├── schemas.py
├── workflow.py
├── runbooks/
│   ├── checkout-database.md
│   ├── identity-access.md
│   └── payment-latency.md
└── tests/
    └── test_retrieval.py

Install the packages:

python -m pip install -U \
  langchain \
  langchain-openai \
  langchain-text-splitters \
  pydantic \
  python-dotenv \
  pytest

The application calls provider embeddings. Tests swap in a small local substitute, so no provider calls happen.

Step 1: write runbooks that can be identified

The checkout runbook starts with a title line:

# Checkout database connection failures

Followed by the service, owner and body:

Service: checkout-api
Owner: commerce-platformUse this runbook when checkout-api reports database connection refused,
pool exhaustion or authentication failures.## First checks1. Compare the active database host, port and database name with the
   previous successful deployment.
2. Confirm that the application secret points to the expected
   credential version.
3. Test network reachability from the application environment without
   printing the credential.
4. Compare connection-pool limits with the database connection limit.

Title, service and owner are plain text, and the loader also copies chosen values into metadata. In a company system, never let a heading define access. Access labels need to originate in a trusted system and get verified whenever retrieval runs.

Step 2: load each file into a Document

Put load_runbooks into retrieval.py. It needs the path helper first:

from pathlib import Path

Then the loader itself:

from langchain_core.documents import Document
def load_runbooks(runbook_dir: Path) -> list[Document]:
    documents: list[Document] = []    for path in sorted(runbook_dir.glob("*.md")):
        text = path.read_text(encoding="utf-8")
        title = text.splitlines()[0].removeprefix("# ").strip()
        service_line = next(
            line for line in text.splitlines()
            if line.startswith("Service:")
        )
        service = service_line.split(":", 1)[1].strip()        documents.append(
            Document(
                page_content=text,
                metadata={
                    "source_id": path.stem,
                    "title": title,
                    "service": service,
                    "path": str(path),
                },
            )
        )    if not documents:
        raise ValueError(f"No Markdown runbooks found in {runbook_dir}")    return documents

Going through it:

glob("*.md") picks the Markdown files in one known directory.
sorted(...) keeps local indexing order repeatable. Search quality shouldn't depend on file-system order, but repeatable input makes debugging easier.
The file gets decoded as UTF-8.
Pulling out the title and service is kept naive on purpose, because the sample files follow one fixed layout. A real loader should validate metadata and report malformed files instead of guessing.
page_content holds the searchable text.
metadata holds fields that should follow every later chunk.
The final check stops an empty index from looking like a success.

Step 3: split the documents

Use RecursiveCharacterTextSplitter:

from langchain_text_splitters import RecursiveCharacterTextSplitter

def split_runbooks(documents: list[Document]) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=420,
        chunk_overlap=60,
        add_start_index=True,
    )
    return splitter.split_documents(documents)

chunk_size=420 asks for chunks of about 420 characters, which suits this small example. chunk_overlap=60 repeats a little text between neighboring chunks, which can keep a sentence next to its explanation when a boundary lands in the middle. More overlap isn't automatically better: it means more stored text, more embedding work and more near-duplicate hits.

With add_start_index=True, each chunk records the character offset where it began, so you can point back into the original file. split_documents hands back fresh Document objects, and every chunk inherits the metadata of its parent.

Picking a chunk size

Copying a number from a tutorial isn't a decision. Look first at how your documents are organized and what people actually ask. A troubleshooting step may need its heading, conditions and warning inside one chunk, and a long policy table may call for a structure-aware splitter. Then test retrieval against labeled questions.

Step 4: create the vector-store retriever

To keep the example small, the app uses InMemoryVectorStore:

from langchain_core.vectorstores import InMemoryVectorStore

def build_vector_retriever(runbook_dir: Path, embeddings, *, k: int = 3):
    chunks = split_runbooks(load_runbooks(runbook_dir))
    store = InMemoryVectorStore.from_documents(
        chunks,
        embedding=embeddings,
    )
    return store.as_retriever(search_kwargs={"k": k})

from_documents embeds and stores the chunks, and as_retriever exposes the standard retriever interface. k=3 requests three documents. That doesn't mean all three are relevant, and it doesn't guarantee the right passage is among them.

The application builds provider embeddings:

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model=os.environ["EMBEDDING_MODEL"]
)
retriever = build_vector_retriever(
    Path(__file__).parent / "runbooks",
    embeddings,
)

Available models and their names depend on your provider account, so keep the model name in configuration. And InMemoryVectorStore loses its index when the process ends. That's fine for a teaching example and small tests, but it's no shared production index.

Step 5: build a focused query from trusted fields

Don't search with a giant dump of the whole application state. Pick selected classification fields:

def build_query(classification: TicketClassification) -> str:
    values = [
        classification.service or "unknown service",
        classification.category,
        classification.summary,
        *classification.evidence,
    ]
    return " | ".join(values)

For the running ticket, the query comes out roughly like this:

checkout-api | database | EU checkout fails after a deployment. |
database connection refused

The string joins the probable service, the problem category, a brief summary and the evidence lifted from the ticket. Secrets, hidden instructions and the full message history stay out of it.

Step 6: carry metadata through the conversion

What the retriever gives back are Document objects, and each one should become a typed passage:

class RetrievedPassage(BaseModel):
    source_id: str
    title: str
    service: str
    start_index: int
    text: str

Then map the Document to it:

def to_passage(document: Document) -> RetrievedPassage:
    return RetrievedPassage(
        source_id=str(document.metadata["source_id"]),
        title=str(document.metadata["title"]),
        service=str(document.metadata["service"]),
        start_index=int(document.metadata.get("start_index", 0)),
        text=document.page_content.strip(),
    )

Required fields are read with square brackets, so a chunk lacking source_id, title or service raises an error right away. start_index gets a default value because chunks stored earlier may not have it, and a production migration should decide if that fallback is acceptable.

Step 7: build a narrow recommendation input

The model-facing writer receives this contract:

class RecommendationInput(BaseModel):
    ticket_id: str
    ticket_summary: str
    category: Category
    severity: Severity
    service: str | None
    evidence: list[str]
    passages: list[RetrievedPassage] = Field(
        min_length=1,
        max_length=4,
    )

When no passages come back, there's no grounded recommendation:

def prepare_recommendation_input(
    classification: TicketClassification,
    passages: list[RetrievedPassage],
) -> RecommendationInput:
    if not passages:
        raise ValueError("At least one runbook passage is required")

    return RecommendationInput(
        ticket_id=classification.ticket_id,
        ticket_summary=classification.summary,
        category=classification.category,
        severity=classification.severity,
        service=classification.service,
        evidence=classification.evidence,
        passages=passages[:4],
    )

If search finds nothing, the model shouldn't be left to improvise from what it knows in general. Route the ticket to a person, or use a fallback that has been approved on its own.

Step 8: ask for source IDs in structured output

The result schema includes citations:

class GroundedRecommendation(BaseModel):
    explanation: str = Field(min_length=20, max_length=500)
    next_checks: list[str] = Field(min_length=1, max_length=4)
    source_ids: list[str] = Field(min_length=1, max_length=4)
    needs_human_review: bool

The prompt reads:

Use only the supplied ticket fields and runbook passages.
Every recommended check must be supported by at least one supplied passage.
Return the source_id values you used.
Do not claim that a check or production change was performed.

Prompt instructions help, but application-level checks still matter.

Step 9: reject citations that retrieval didn't return

Once the writer has run, compare its source IDs with the retrieved passages:

available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available

if unknown:
    raise ValueError(
        "Recommendation cited sources that retrieval did not return: "
        + ", ".join(sorted(unknown))
    )

This catches a fabricated source ID and nothing more. A model can cite a document that really was returned and still get its meaning wrong, so evaluation and human review are still required.

Step 10: connect the fixed workflow

The request-time code is plain:

def run_retrieval_workflow(
    classification: TicketClassification,
    retriever,
    recommendation_chain,
) -> RetrievalResult:
    query = build_query(classification)
    passages = retrieve_passages(retriever, query)
    values = prepare_recommendation_input(classification, passages)
    recommendation = recommendation_chain.invoke(values)

    available = {passage.source_id for passage in passages}
    unknown = set(recommendation.source_ids) - available
    if unknown:
        raise ValueError("Recommendation cited an unavailable source")    return RetrievalResult(
        query=query,
        recommendation=recommendation,
        passages=passages,
    )

There's no hidden loop. Search runs once, the writer runs once, and the source check runs once.

What comes out

A result looks like this:

{
  "query": "checkout-api | database | EU checkout fails after a deployment. | database connection refused",
  "recommendation": {
    "explanation": "The error matches the checkout database connection runbook.",
    "next_checks": [
      "Compare the active database host, port and database name with the previous successful deployment.",
      "Confirm that the application secret points to the expected credential version."
    ],
    "source_ids": ["checkout-database"],
    "needs_human_review": false
  },
  "passages": [
    {
      "source_id": "checkout-database",
      "title": "Checkout database connection failures",
      "service": "checkout-api",
      "start_index": 0
    }
  ]
}

The API can turn each source ID into a link, but only after verifying that the current caller may open it.

Testing retrieval without paying for embeddings

The companion snapshot includes KeywordEmbeddings, a small test substitute that counts a fixed vocabulary:

VOCABULARY = (
    "database",
    "connection",
    "checkout",
    "deployment",
    "credential",
    "pool",
    "payment",
    "latency",
    "timeout",
    "identity",
    "access",
    "role",
    "claim",
)

Don't mistake it for a production embedder. Its only purpose is to keep tests fast, offline and deterministic. The first test checks the running ticket:

passages = retrieve_passages(
    retriever,
    build_query(classification()),
)

assert passages[0].source_id == "checkout-database"
assert passages[0].service == "checkout-api"
assert passages[0].start_index >= 0

The second measures hit at two across three labeled queries:

score = hit_at_k(
    retriever,
    [
        ("checkout database connection refused", "checkout-database"),
        ("payment provider latency timeout", "payment-latency"),
        ("identity role claim access denied", "identity-access"),
    ],
)

assert score == 1.0

Hit at two poses a small question: is the expected source within the top two results? A real test set needs many questions, hard negatives, similar documents and current access rules. A perfect score on three examples proves only that those three pass.

The third test feeds in a fake recommendation citing not-returned, and the workflow has to reject it. Run everything with:

pytest -q

The expected result:

3 passed

Naming the ways retrieval fails

Each failure type deserves its own name, so you can tell where to look.

Indexing failure: a file couldn't be read, metadata was missing or an embedding request failed. Never publish a half-built index as complete.
Freshness failure: a policy changed but search still served the previous runbook. Record versions and timestamps, and decide how edits and deletions propagate into the index.
Permission failure: the index surfaced a passage this caller isn't entitled to read. Apply access filters before any text reaches the model, since hiding the link afterward comes too late.
Ranking failure: the correct document is indexed yet missing from the top hits. Adjust the query, filters, chunking, embeddings or ranking, and measure once more.
Generation failure: retrieval worked, but the writer skipped or misinterpreted the passage. Evaluate the two stages separately to see which one broke.
Citation failure: the answer points to a source that doesn't exist, or to one that doesn't back the claim. Checking IDs against the returned set detects only the first problem.

Retrieved text is untrusted

Runbooks contain errors, and sometimes lines that read like orders to the model. Treat retrieved passages as quoted source material and never as application rules with higher priority. Keep stable behavior in the system prompt, delimit the retrieved text, and restrict tools in application code. No runbook should ever be able to authorize itself to trigger a tool that writes.

When vector search is the wrong tool

Some lookups don't need vectors. An exact ticket ID belongs in a plain database query, and a known service name maps straight to a catalog entry. Exact error codes and rare identifiers are better served by keyword search. If the company already runs a search service that indexes the right documents and enforces permissions, use that. Vector search is only one tool among several.

What's still missing for production

Several things are still absent before a company could rely on this path:

a persistent store instead of memory
someone clearly owning every document
access filters applied at query time
handling for document versions and deletions
a plan for the day the embedding model changes
a cap on how much text comes back
labeled tests for retrieval
generation tests anchored in the retrieved passages
records of cost and latency
audit logs that keep secrets out

The framework connects the parts, and the application owns those decisions.

Where the helpdesk app stands

It can now search a controlled runbook set and keep its sources:

classification
  -> focused query
  -> retriever
  -> typed passages with metadata
  -> recommendation
  -> source-ID check

What comes next is splitting message history, saved user facts, live workflow state and checkpoints into separate things. Drawing that line also shows whether a plain LangChain chain still suffices or LangGraph fits better.