All docs
Examples
Hosted workflow

RAG policy assistant workflow

Make policy retrieval tenant-safe, versioned, evidenced, and replayable instead of prompt-only.

Problem

RAG workflows can retrieve chunks, but they may not prove the chunks were allowed, current, retained, and attached to the answer.

Why app glue gets messy

The app usually owns retrieval filters, citation coverage, policy versions, and replay reconstruction.

How Synapsor helps

Synapsor searches only allowed policy chunks, records chunk ids/versions in evidence, and retains the run for later replay.

Value sources

Example names and seed ids are developer-defined. SESSION values are set by your backend. ARG values come from the SDK/HTTP call. Handles such as wrp://..., evidence://..., and agent-run://... are returned by Synapsor and should be stored for audit/replay.

Read the value-source guide
Expected outcome

A policy answer can show the exact chunk versions used months after the policy changes.

Production checks

  • Tenant and policy_domain filters are bound from session.
  • Evidence includes policy_chunk ids and versions.
  • Old runs replay against the stored evidence rather than current policy text.
  • Answers without citation coverage are rejected or held.
schema.sql
CREATE TABLE policy_chunks (
  id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  domain TEXT NOT NULL,
  version TEXT NOT NULL,
  body TEXT NOT NULL
);
agent.ddl.sql
CREATE AGENT WORKFLOW policy.answer_policy_flow
SESSION REQUIRE tenant_id, principal, policy_domain
ALLOWED CAPABILITIES (
  policy.answer_question,
  policy.compare_policy_versions,
  policy.explain_applicable_rule
)
EVIDENCE REQUIRED
CHECKPOINT EVERY STEP
RETENTION RUNS 365 DAYS, EVIDENCE 365 DAYS;
python
run = db.agent_runs.start(
    workflow="policy.answer_policy_flow",
    input={"question": "Can we waive this late fee?"},
)

answer = run.invoke_capability(
    "policy.answer_question",
    step_key="answer_policy_question",
    arguments={"question": "Can we waive this late fee?"},
    response_envelope=True,
)
print(answer["evidence_bundle_id"])