Parent vs child graph

A graph of graphs sounds abstract until you see why we need it. The parent graph orchestrates the end-to-end flow: anonymize, rewrite, retrieve, synthesize, evaluate. The child graph is the retriever: embed, search, optionally filter. The parent calls the compiled child once per rewritten query. Splitting them means we can swap the retriever for a hybrid or reranked variant without touching the parent.

rag_graph.py
python
class SubState(TypedDict, total=False):
    """State for the child retriever graph (one rewritten query)."""
    query: str
    top_k: int
    embedding: list[float]
    chunks: list[DocumentChunk]


class GraphState(TypedDict, total=False):
    """State for the parent graph."""
    question: str                     # original user question
    anonymize: bool
    anonymized_question: str
    pii_mapping: dict[str, str]
    rewrite_n: int
    rewrites: list[str]
    top_k: int
    retrieved: list[DocumentChunk]    # deduped union from all sub-retrievals
    contexts: list[str]               # content strings for synthesizer + eval
    answer: str
    with_evaluation: bool
    eval_score: dict | None

Separate TypedDicts for separate scopes. The child state carries one query at a time. The parent state carries the full run including rewrites, fused chunks, the final answer, and optional eval scores.

rag_graph.py
python
def build_parent_graph():
    g = StateGraph(GraphState)
    g.add_node("anonymize", _anonymize_node)
    g.add_node("rewrite", _rewrite_node)
    g.add_node("sub_retrieve", _sub_retrieve_node)
    g.add_node("synthesize", _synthesize_node)
    g.add_node("evaluate", _evaluate_node)

    g.set_entry_point("anonymize")
    g.add_edge("anonymize", "rewrite")
    g.add_edge("rewrite", "sub_retrieve")
    g.add_edge("sub_retrieve", "synthesize")
    g.add_edge("synthesize", "evaluate")
    g.add_edge("evaluate", END)
    return g.compile()


# Compile once at import so the first request isn't penalized.
parent_graph = build_parent_graph()

The parent graph is linear. Every node reads from and writes to GraphState. Compiling at import time means the first request is not slower than the rest.

rag_graph.py
python
async def run_graph(
    question: str,
    rewrite_n: int = 3,
    top_k: int = 5,
    anonymize: bool = True,
    with_evaluation: bool = False,
) -> Dict[str, Any]:
    """Invoke the compiled parent graph with a fresh state dict."""
    initial: GraphState = {
        "question": question,
        "anonymize": anonymize,
        "rewrite_n": rewrite_n,
        "top_k": top_k,
        "with_evaluation": with_evaluation,
    }
    result = await parent_graph.ainvoke(initial)
    return {
        "answer": result.get("answer", ""),
        "rewrites": result.get("rewrites", []),
        "sources": [
            {
                "source": c.source,
                "chunk_index": c.chunk_index,
                "content": c.content[:300] + ("..." if len(c.content) > 300 else ""),
            }
            for c in (result.get("retrieved") or [])
        ],
        "eval_score": result.get("eval_score"),
    }

run_graph is the service-facing wrapper around parent_graph.ainvoke. It builds a fresh GraphState, runs the graph, and reshapes the result for the API: the deduped retrieved chunks become the sources array (truncated to 300 characters each), alongside answer, rewrites, and eval_score. Every /ask call you have made so far went through this function, which is why the response shows sources even though the state key is named retrieved.

Quiz: Quiz

Loading practice…