What you will build
Our technical branch currently returns generic advice. We now need it to consult a small collection of workshop notes before producing a reply. The question, source notes, and citation rules will all be visible in the file. There is no database server, vector service, or language model to configure.
We will also introduce a failure path. If a draft does not pass the structural citation check, the graph allows one revision and then escalates. The stopping rule is as important as the successful example: an automated workflow should not repeat forever.
By the end: Retrieve a local source note, prepare a cited draft, and bound the revision loop.
- What you will build
- What you need before starting
- Understand the idea before the code
- The workflow we are building
- Create the lesson folder
- Complete code for this article
- Run it and compare the result
- Walk through what just happened
- Change one thing yourself
- If your result is different
- What you can now explain
What you need before starting
You need basic Python: variables, functions, dictionaries, loops, and imports. When this lesson introduces a class or a library-specific term, its role is explained below.
Use Python 3.12. These examples were executed on Windows with Python 3.12.4. Use the shown interpreter consistently when installing packages and running code.
You do not need to run an earlier article to make this part execute. The complete files and data used here are supplied below. The series order is for learning the ideas progressively.
Understand the idea before the code
The workflow now needs a relevant workshop note before drafting a reply. Retrieval means searching an existing collection and returning matching material. It does not train a model. Our collection is a small local set of notes with a source ID, title, and text.
An evidence item is one retrieved note supplied to the drafting step. A source ID is a stable identifier such as a note's ID; a citation is the reply's reference to such an ID. Keeping those IDs lets the program check whether the reply refers to material that was actually provided.
The state schema, TicketState, describes the dictionary fields used by this larger ticket workflow. initial_state(...) creates the actual starting values for a particular ticket. The schema is not the note library; the state holds the current question and, after lookup, the retrieved evidence.
Our retrieval function uses simple word overlap. If a question contains words that occur in a suitable note, that note can receive a higher match score. A score is a ranking signal in this small rule-based search, not the probability that the answer is true. There is no vector database or embedding model hidden in the lesson. An embedding, in other systems, is a learned numeric representation used to compare meaning; we do not require one here.
The workflow we are building
The boxes are named Python operations. Solid arrows are fixed transitions; dashed arrows show possible destinations chosen by routing functions. The image comes from this lesson's actual compiled graph. One particular input follows only the applicable route. Open the full-size graph.
Create the lesson folder
Create this folder and keep the filenames exactly as shown. In particular, a local module is an ordinary file beside the program, not an extra package you must guess how to install.
lesson-folder/
lesson.py
graph_picture.py
requirements.txt
Save requirements.txt
These are the external Python packages for this lesson. The local files above are included in the article; they are not pip packages.
langgraph==1.2.16
langgraph-checkpoint-sqlite==3.1.2
python -m venv .venv
.venv\Scripts\python.exe -m pip install -r requirements.txt
On Linux/macOS, use python3 -m venv .venv and .venv/bin/python -m pip install -r requirements.txt. Installation requires internet access. The default examples do not call a paid API.
Complete code for this article
Save this complete file as lesson.py
This is the complete program, including imports, data, function definitions, and the entry point.
from graph_picture import save_graph
from operator import add
from typing import Annotated, Literal, TypedDict
Category = Literal["setup", "hardware", "billing", "other"]
class Evidence(TypedDict):
id: str
title: str
text: str
class TicketState(TypedDict, total=False):
ticket_id: str
text: str
category: Category
evidence: list[Evidence]
reply: str
citations: list[str]
attempts: int
checks_ok: bool
approved: bool
outcome: str
llm_used: bool
events: Annotated[list[str], add]
def initial_state(ticket_id: str, text: str) -> TicketState:
if not ticket_id or len(ticket_id) > 80:
raise ValueError("Choose a ticket ID containing 1 to 80 characters.")
if not text.strip() or len(text) > 4000:
raise ValueError("Ticket text must contain 1 to 4,000 characters.")
return {
"ticket_id": ticket_id,
"text": text.strip(),
"attempts": 0,
"events": [],
"approved": False,
"llm_used": False,
}
KNOWLEDGE = [
{
"id": "D1",
"category": "setup",
"title": "Create an isolated Python environment",
"keywords": ["venv", "virtual", "environment", "activate", "python"],
"text": "Create a virtual environment with python -m venv .venv. Use the Python executable inside .venv when installing and running this project.",
},
{
"id": "D2",
"category": "setup",
"title": "Install into the correct interpreter",
"keywords": [
"pip",
"install",
"module",
"import",
"modulenotfounderror",
"package",
],
"text": "Check python -m pip --version to identify the interpreter used for package installation. Install the required package with that same interpreter.",
},
{
"id": "D3",
"category": "hardware",
"title": "Investigate an unavailable serial port",
"keywords": ["esp32", "port", "serial", "usb", "cable", "board"],
"text": "For this fictional workshop, first identify the exact board and its vendor instructions. Check that the USB cable supports data, the expected device appears in your operating system, and the selected port matches it. Disconnect power before changing wiring.",
},
]
import json
import re
def tokens(text: str) -> set[str]:
return set(re.findall("[a-z0-9]+", text.lower()))
def classify_rules(text: str) -> str:
words = tokens(text)
if words & {"refund", "payment", "invoice", "charge", "billing"}:
return "billing"
if words & {"esp32", "port", "serial", "usb", "board", "camera"}:
return "hardware"
if words & {"python", "pip", "venv", "install", "module", "import", "environment"}:
return "setup"
return "other"
def retrieve(text: str, category: str) -> list[dict]:
corpus = KNOWLEDGE
words = tokens(text)
scored = [
(len(words & set(doc["keywords"])), doc)
for doc in corpus
if doc["category"] == category
]
scored.sort(key=lambda pair: (-pair[0], pair[1]["id"]))
return [
{key: doc[key] for key in ("id", "title", "text")}
for score, doc in scored
if score > 0
][:2]
import re
def draft_demo(state: dict) -> dict:
evidence = state.get("evidence", [])
if not evidence:
return {
"reply": "A workshop mentor needs to review this question.",
"citations": [],
"llm_used": False,
}
doc = evidence[0]
return {
"reply": f'Try this documented starting point: {doc['text']} [{doc['id']}]',
"citations": [doc["id"]],
"llm_used": False,
}
def citation_checks(state: dict) -> bool:
"""Check IDs and markers. This does NOT verify factual entailment."""
citations = state.get("citations", [])
reply = state.get("reply", "")
allowed = {doc["id"] for doc in state.get("evidence", [])}
if not citations or not reply.strip():
return False
if len(citations) != len(set(citations)):
return False
markers = set(re.findall("\\[(D[0-9]+)\\]", reply))
return set(citations) <= allowed and markers == set(citations)
"Article 4: retrieve evidence and bound the draft-check loop."
import json
from langgraph.graph import END, START, StateGraph
def build(drafter=draft_demo):
def classify(state):
return {"category": classify_rules(state["text"])}
def route(state):
return "lookup" if state["category"] in {"setup", "hardware"} else "human"
def lookup(state):
return {"evidence": retrieve(state["text"], state["category"])}
def after_lookup(state):
return "draft" if state["evidence"] else "human"
def check(state):
return {"checks_ok": citation_checks(state)}
def after_check(state):
if state["checks_ok"]:
return END
return "revise" if state["attempts"] < 1 else "human"
def revise(state):
return {"attempts": state["attempts"] + 1}
def human(state):
return {"outcome": "needs_human"}
builder = StateGraph(TicketState)
for name, fn in [
("classify", classify),
("lookup", lookup),
("draft", drafter),
("check", check),
("revise", revise),
("human", human),
]:
builder.add_node(name, fn)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, ["lookup", "human"])
builder.add_conditional_edges("lookup", after_lookup, ["draft", "human"])
builder.add_edge("draft", "check")
builder.add_conditional_edges("check", after_check, [END, "revise", "human"])
builder.add_edge("revise", "draft")
builder.add_edge("human", END)
compiled = builder.compile()
save_graph(compiled, 4)
return compiled
if __name__ == "__main__":
print(
json.dumps(
build().invoke(
initial_state("lesson-4", "My pip install has a missing module")
),
indent=2,
)
)
Save this complete file as graph_picture.py
This is the complete local diagram renderer imported by lesson.py. Save it in the same folder. It draws the compiled graph; it does not decide the workflow or contact an image service.
import html
from pathlib import Path
ROLES = {
"__start__": "Enter the workflow",
"__end__": "This run is finished",
"clean": "Normalize the question",
"draft": "Prepare a reply",
"classify": "Choose a topic category",
"technical": "Technical branch",
"human": "Ask a mentor to review",
"lookup": "Retrieve local evidence",
"check": "Check source-ID structure",
"revise": "Allow one more draft",
"review": "Pause inside this node",
"approval": "Wait for a human decision",
"queue": "Write one local outbox row",
"escalate": "Finish with needs_human",
}
def positions(part):
if part == 2:
return {
"__start__": (320, 115),
"clean": (320, 250),
"draft": (320, 385),
"__end__": (320, 520),
}
if part == 3:
return {
"__start__": (420, 115),
"classify": (420, 250),
"human": (120, 410),
"technical": (720, 410),
"__end__": (420, 590),
}
if part == 5:
return {"__start__": (320, 115), "review": (320, 265), "__end__": (320, 415)}
return {
"__start__": (420, 115),
"classify": (420, 250),
"lookup": (420, 385),
"draft": (420, 520),
"check": (420, 655),
"revise": (790, 655),
"human": (70, 850),
"escalate": (70, 850),
"approval": (420, 850),
"queue": (420, 1035),
"__end__": (420, 1215),
}
def draw_svg(structure, part):
layout = positions(part)
node_width, node_height = (225, 83)
width = 850 if part in {2, 5} else 1080
height = max((layout[name][1] for name in structure.nodes)) + node_height + 100
chunks = [
f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" viewBox="0 0 {width} {height}" role="img" aria-labelledby="title desc">',
f'<title id="title">Part {part}: graph generated from the compiled lesson code</title>',
'<desc id="desc">Solid arrows are fixed edges. Dashed arrows are conditional destinations. Node positions are chosen for readability.</desc>',
'<defs><marker id="solid" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="8" markerHeight="8" orient="auto"><path d="M0 0 L10 5 L0 10z" fill="#526c75"/></marker><marker id="conditional" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="8" markerHeight="8" orient="auto"><path d="M0 0 L10 5 L0 10z" fill="#b54625"/></marker></defs>',
'<rect width="100%" height="100%" fill="#faf9f5"/>',
'<text x="30" y="35" font-family="Arial,sans-serif" font-size="16" font-weight="bold" fill="#b54625">CODETIPS ACADEMY / GENERATED FROM CODE</text>',
f'<text x="30" y="73" font-family="Arial,sans-serif" font-size="28" font-weight="bold" fill="#183443">Part {part} — the compiled workflow</text>',
]
edge_data = []
for index, edge in enumerate(structure.edges):
sx, sy = layout[edge.source]
tx, ty = layout[edge.target]
if sx == tx:
points = [
(sx + node_width / 2, sy + node_height),
(tx + node_width / 2, ty),
]
elif ty < sy:
points = [
(sx + node_width / 2, sy),
(sx + node_width / 2, ty + node_height / 2),
(tx + node_width, ty + node_height / 2),
]
else:
leftward = tx < sx
x1 = sx if leftward else sx + node_width
x2 = tx + node_width if leftward else tx
y1 = sy + node_height / 2
y2 = ty + node_height / 2
middle = (x1 + x2) / 2 + (index % 3 - 1) * 12
points = [(x1, y1), (middle, y1), (middle, y2), (x2, y2)]
kind = "conditional" if edge.conditional else "solid"
color = "#b54625" if edge.conditional else "#526c75"
dash = ' stroke-dasharray="9 6"' if edge.conditional else ""
chunks.append(
'<polyline points="'
+ " ".join((f"{x:g},{y:g}" for x, y in points))
+ f'" fill="none" stroke="{color}" stroke-width="3"{dash} marker-end="url(#{kind})"/>'
)
edge_data.append(
{
"source": edge.source,
"target": edge.target,
"conditional": bool(edge.conditional),
}
)
for name in structure.nodes:
x, y = layout[name]
display = {"__start__": "START", "__end__": "END"}.get(name, name)
highlight = name in {"review", "approval", "check"}
chunks.append(
f'<rect x="{x}" y="{y}" width="{node_width}" height="{node_height}" rx="12" fill="{('#f8e9df' if highlight else '#eaf0eb')}" stroke="#c9d6ce" stroke-width="2"/>'
)
chunks.append(
f'<text x="{x + 14}" y="{y + 32}" font-family="Consolas,monospace" font-size="23" font-weight="bold" fill="#183443">{html.escape(display)}</text>'
)
chunks.append(
f'<text x="{x + 14}" y="{y + 60}" font-family="Arial,sans-serif" font-size="15" fill="#526c75">{html.escape(ROLES.get(name, name))}</text>'
)
chunks += [
f'<path d="M30 {height - 55} H{width - 30}" stroke="#d9dfd8"/>',
f'<text x="30" y="{height - 25}" font-family="Arial,sans-serif" font-size="16" fill="#526c75">Solid: fixed edge. Orange dashed: conditional destination. See the routing explanation below.</text>',
"</svg>",
]
return (
"\n".join(chunks),
{"part": part, "nodes": list(structure.nodes), "edges": edge_data},
)
def save_graph(graph, part):
structure = graph.get_graph()
svg, _ = draw_svg(structure, part)
Path("workflow.svg").write_text(svg, encoding="utf-8")
Path("workflow.mmd").write_text(structure.draw_mermaid(), encoding="utf-8")
Run it and compare the result
Run commands from the lesson folder. In the commands below, python means the environment you installed into. On Windows, substitute .venv\Scripts\python.exe; on Linux/macOS substitute .venv/bin/python. Do not create the environment with one interpreter and run the lesson with another.
Run 1: python lesson.py
python lesson.py
Captured standard output:
{
"ticket_id": "lesson-4",
"text": "My pip install has a missing module",
"category": "setup",
"evidence": [
{
"id": "D2",
"title": "Install into the correct interpreter",
"text": "Check python -m pip --version to identify the interpreter used for package installation. Install the required package with that same interpreter."
}
],
"reply": "Try this documented starting point: Check python -m pip --version to identify the interpreter used for package installation. Install the required package with that same interpreter. [D2]",
"citations": [
"D2"
],
"attempts": 0,
"checks_ok": true,
"approved": false,
"llm_used": false,
"events": []
}
The program also writes workflow.svg and workflow.mmd into the current folder. Open the SVG in a browser. The Mermaid file is a text representation of the same nodes and edges.
The displayed output was captured from the supplied default program in a separate folder, with no project-level module on its import path. It is a reproducible demonstration on these examples, not a claim about arbitrary inputs.
Walk through what just happened
1. Keep source data visible
KNOWLEDGE contains the complete three-note dataset. retrieve compares question words with each note’s keywords and returns up to two matching notes in the selected category. It is a simple ranking rule, not semantic search.
2. Separate evidence from the reply
lookup adds evidence to the ticket state. draft_demo uses one retrieved note and returns both reply text and the IDs it cites. The state lets a reviewer inspect the actual evidence behind that draft.
3. Check only what the code can check
citation_checks rejects missing, duplicate, or unavailable source IDs and mismatched citation markers. It does not prove that the reply interprets the source correctly. A structurally valid citation is not a fact-check.
4. Trace the bounded loop
attempts starts at zero. A failed check can visit revise once, which increments it to one and returns to draft. A second failure exits to human. A successful check ends this lesson’s graph; it does not queue or send a message.
Change one thing yourself
Temporarily make draft_demo return an empty citations list. Run the program and follow attempts and outcome. Restore the code after observing the needs_human exit.
If your result is different
If no evidence is returned, inspect the category and keyword overlap before changing the graph. If the loop never stops, check both the counter update and the condition after check.
Check the complete imports and filenames first. Then compare your input and command with the shown run. Numeric results may differ slightly between platforms; the program's assertions check the intended relationships rather than relying on a decorative screenshot.
What you can now explain
Retrieve a local source note, prepare a cited draft, and bound the revision loop. Next we will pause a workflow for a reviewer and save enough progress to resume after Python exits.
Technical reference: LangGraph Graph API. This project and its wording are original CodeTips Academy material.
