Repository
⌥ github.com/WillCatt/PolicyDesk ↗
Python · LangGraph · PydanticAI · langchain-ollama / -anthropic · rank-bm25 · Ollama (llama-3.1-8b) · pytest. Reuses the QuoteGuard PDS corpus + retrieval approach; the new layer is the orchestration.
Structure
src/policydesk/
├── graph.py StateGraph: fan-out + critic⇄revise loop + checkpointer (+ optional HITL)
├── state.py DeskState (typed) + reducers for the fan-out merge
├── nodes/ guarding (+human_approval) · planner · research · critic (+revise) · synthesizer
├── router.py complexity classifier → single-shot or the desk
├── planner_pydantic.py the planner as a typed PydanticAI agent
├── retrieval.py BM25 over the 138 PDS chunks
├── models.py ChatOllama / ChatAnthropic factory + usage capture
├── guards.py input refusals + output grounding/citation checks
└── baseline.py single-shot RAG (the control)
eval/ run.py · router_eval.py · pydantic_compare.py · export_demo.py (all with bootstrap CIs)
examples/ hitl_demo.py · figures/ build_portfolio_figures.py · tests/ (run without Ollama)
The graph, in one function
The whole orchestration is declarative: nodes, a Send fan-out from the planner, and a conditional edge that forms the critic ⇄ revise cycle.
def build_graph(checkpointer=None):
g = StateGraph(DeskState)
for name, fn in [("guard_in", guard_in), ("planner", planner), ("research", research),
("critic", critic), ("revise", revise),
("synthesizer", synthesizer), ("guard_out", guard_out)]:
g.add_node(name, fn)
g.add_edge(START, "guard_in")
g.add_conditional_edges("guard_in", _after_guard_in, {"blocked": END, "ok": "planner"})
g.add_conditional_edges("planner", _fan_out_research, ["research"]) # Send map-reduce
g.add_edge("research", "critic") # fan back in
g.add_conditional_edges("critic", _after_critic, {"revise": "revise", "synthesize": "synthesizer"})
g.add_edge("revise", "critic") # the cycle
g.add_edge("synthesizer", "guard_out")
g.add_edge("guard_out", END)
return g.compile(checkpointer=checkpointer or MemorySaver())
Fan-out + the capped loop
def _fan_out_research(state): # one researcher per sub-question
return [Send("research", {"question": state["question"], "subquestion": sq})
for sq in state["subquestions"]]
def _after_critic(state): # loop until clean or the cap is hit
cap = state.get("max_revisions", settings.max_revisions)
flagged = any(c["verdict"] == "revise" for c in state.get("critiques", []))
return "revise" if flagged and state.get("revision_count", 0) < cap else "synthesize"
The critic: deterministic check overrides the LLM
A cheap, exact grounding check runs first; the LLM faithfulness verdict can't rescue a draft that cites a page it never retrieved.
g = check_grounding(draft, evidence_pages) # cited ⊆ retrieved, and ≥1 citation
if not g.grounded:
verdict = "revise" # deterministic override
reason = "No page citation." if not g.has_citation else f"Cites pages not in evidence {g.unsupported_pages}."
else:
verdict = "revise" if llm_verdict == "revise" else "pass"
Routing by complexity (zero-cost classifier)
A clause after an "and" that carries its own interrogative head marks a compound question; simple ones skip the desk.
_HEADS = r"\b(what|which|how|when|where|who|does|do|is|are|can|must|will|until|within)\b"
def classify_complexity(question: str) -> str:
parts = re.split(r",?\s+and\s+|\s*;\s*", question.lower())
if len(parts) >= 2 and any(re.search(_HEADS, p) for p in parts[1:]):
return "compound"
return "simple"
def route_answer(q):
return run_baseline(q) if classify_complexity(q) == "simple" else run_desk(q)
The planner on PydanticAI (typed, validated)
The same job as the LangGraph planner node, but the output is a validated Plan with retries — no hand-rolled JSON parsing or fallback.
class Plan(BaseModel):
subquestions: list[str] = Field(..., min_length=1)
agent = Agent(OpenAIChatModel(settings.ollama_model, provider=OpenAIProvider(base_url=OLLAMA_V1, api_key="ollama")),
output_type=PromptedOutput(Plan), # schema-in-prompt; 8B models handle this, not tool-calls
system_prompt=PLANNER_SYSTEM, retries=3)
plan: Plan = agent.run_sync(f"User question: {q}").output # typed + validated, or raises after retries
Human-in-the-loop: pause, approve, resume
One LangGraph interrupt turns the graph into an approval workflow; the checkpointer holds the paused state between calls.
def human_approval(state): # node wired in by build_graph(human_in_loop=True)
review = interrupt({"brief": state["final_brief"], "subquestions": state["subquestions"]})
return {} if review.get("approved") else {"final_brief": WITHHELD, "status": "withheld"}
result = graph.invoke(inputs, cfg) # ... runs until the interrupt, then pauses
graph.invoke(Command(resume={"approved": True}), cfg) # reviewer resumes from the checkpoint
