["NEVER answer without citing sources","If no relevant sources found, say 'I cannot find relevant policy guidance'","Include section references and document URLs in citations","Flag uncertainty: if confidence < 0.8, say 'I'm not certain...'","No speculation or inference beyond what's in documents"]
inputs
[{"question":"Policy question from user"},{"top_k":"Number of sources to retrieve (default: 5)"},{"min_confidence":"Minimum confidence score (default: 0.8)"}]
workflow
[{"step":"Embed question using same model as policy corpus"},{"step":"Retrieve top_k similar chunks from pgvector"},{"step":"Rank by relevance score"},{"step":"If max score < min_confidence, refuse to answer"},{"step":"Generate answer citing specific sections and documents"},{"step":"Include all source URLs and section references"},{"step":"Log query, sources, answer in audit trail"}]
success_criteria
["Answers include citations with section references","Refuses to answer when sources missing or low confidence","Citation precision ≥0.95 (citations actually support claims)","All interactions logged for audit"]
Policy Q&A Skill
Purpose
Answer policy and SOP questions with strict source attribution. Refuses to answer without relevant sources. Designed for compliance-sensitive environments.
Usage
# Ask policy question
answer = policy_qa(
question="What is the approval workflow for journal entries over $100K?",
top_k=5,
min_confidence=0.8
)
Workflow
1. Embed Question
from sentence_transformers import SentenceTransformer
defembed_question(question, model_name='sentence-transformers/all-MiniLM-L6-v2'):
"""Embed question using same model as policy corpus"""
model = SentenceTransformer(model_name)
embedding = model.encode(question)
return embedding.tolist()
defrank_sources(sources, question):
"""Re-rank sources by relevance using cross-encoder"""from sentence_transformers import CrossEncoder
model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
# Score each source
pairs = [[question, s['text']] for s in sources]
scores = model.predict(pairs)
# Add scores to sourcesfor source, score inzip(sources, scores):
source['relevance_score'] = float(score)
# Sort by relevance
sources.sort(key=lambda x: x['relevance_score'], reverse=True)
return sources
4. Check Confidence Threshold
defcheck_confidence(sources, min_confidence=0.8):
"""Check if top source meets confidence threshold"""ifnot sources:
returnFalse, "No relevant sources found"
top_score = sources[0]['relevance_score']
if top_score < min_confidence:
returnFalse, f"Low confidence ({top_score:.2f} < {min_confidence})"returnTrue, None
5. Generate Answer with Citations
defgenerate_answer_with_citations(question, sources):
"""Generate answer citing specific sources"""# Build context from top sources
context = "\n\n".join([
f"[{i+1}] {s['document_name']}, Section: {s['section']}, Page: {s['page_number']}\n{s['text']}"for i, s inenumerate(sources[:3]) # Use top 3 sources
])
# Prompt for answer generation
prompt = f"""
Answer the following question based ONLY on the provided policy documents.
You MUST cite your sources using [1], [2], [3] notation.
Question: {question}
Policy Documents:
{context}
Instructions:
- Answer the question using only information from the provided documents
- Cite sources inline using [1], [2], [3]
- If the documents don't contain the answer, say "I cannot find relevant policy guidance"
- Be specific about policy requirements, procedures, approvals
- Quote exact text when citing requirements
Answer:
"""# Generate answer (using LLM)
answer = call_llm(prompt)
# Verify citations are presentifnotany(f'[{i+1}]'in answer for i inrange(len(sources[:3]))):
return"I cannot provide an answer without proper citations."return answer
defcall_llm(prompt):
"""Call LLM API (e.g., Claude, GPT, or local model)"""# Implementation depends on LLM choicepass
# tests/finance/policy-qa.yamlsuite:policy-qathresholds:citation_precision:0.95refuses_without_source:truecases:-id:ev-journal-approvalprompt:"What is the approval workflow for journal entries over $100K?"expects:-has_citations:true-citation_count: [1, 2, 3] # 1-3 citations-cites_correct_policy:true-includes_source_urls:true-id:ev-low-confidence-refusalprompt:"What is the company's policy on flying cars?"expects:-refuses_to_answer:true-mentions_no_sources:true
Policy Corpus Ingestion
Ingest policy documents into pgvector:
defingest_policy_document(pdf_path, document_name, url):
"""Ingest policy PDF into pgvector"""# 1. Extract text from PDFfrom pypdf import PdfReader
reader = PdfReader(pdf_path)
sections = []
for page_num, page inenumerate(reader.pages, start=1):
text = page.extract_text()
# Chunk by section headings or fixed size
chunks = chunk_text(text, chunk_size=500, overlap=100)
for chunk in chunks:
sections.append({
'text': chunk,
'document_name': document_name,
'page_number': page_num,
'section': detect_section(chunk), # Extract section heading'url': url,
})
# 2. Embed all chunks
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
texts = [s['text'] for s in sections]
embeddings = model.encode(texts)
# 3. Store in pgvector
conn = psycopg2.connect(os.environ['POSTGRES_URL'])
cursor = conn.cursor()
for section, embedding inzip(sections, embeddings):
cursor.execute("""
INSERT INTO policy_chunks (
text, document_name, section, page_number, url, embedding
) VALUES (%s, %s, %s, %s, %s, %s)
""", (
section['text'],
section['document_name'],
section['section'],
section['page_number'],
section['url'],
embedding.tolist(),
))
conn.commit()
cursor.close()
conn.close()
Schema
CREATE TABLE policy_chunks (
id SERIAL PRIMARY KEY,
text TEXT NOT NULL,
document_name VARCHAR(255) NOT NULL,
section VARCHAR(255),
page_number INTEGER,
url TEXT,
embedding vector(384), -- MiniLM dimension
created_at TIMESTAMPDEFAULT NOW()
);
CREATE INDEX ON policy_chunks USING ivfflat (embedding vector_cosine_ops);
CREATE INDEX ON policy_chunks (document_name, section);
CREATE TABLE policy_qa_audit_log (
id SERIAL PRIMARY KEY,
user_id VARCHAR(255) NOT NULL,
timestampTIMESTAMPNOT NULL,
question TEXT NOT NULL,
sources JSONB, -- Array of source IDs
answer TEXT,
refused BOOLEANNOT NULL,
confidence_score FLOAT,
created_at TIMESTAMPDEFAULT NOW()
);
CREATE INDEX ON policy_qa_audit_log (user_id, timestamp);
CREATE INDEX ON policy_qa_audit_log (refused);
Guardrails
Never Answer Without Sources
ifnot sources or sources[0]['relevance_score'] < min_confidence:
return"I cannot find relevant policy guidance on this topic. Please consult the full policy library or contact Compliance."
Flag Uncertainty
if sources[0]['relevance_score'] < 0.9:
answer = f"⚠️ **Moderate confidence** ({sources[0]['relevance_score']:.2f})\n\n{answer}"
No Speculation
Prompt includes:
- Do NOT infer or speculate beyond what's explicitly stated
- If the policy doesn't address this specific case, say so
- Quote exact text when citing requirements
