-
Notifications
You must be signed in to change notification settings - Fork 1
Project Plan
Stages: All four stages in sequence. Stages 2 and 3 are fully AI-powered with RAG. Stages 1 and 4 are human-input placeholder gates — they exist as real workflow steps with structured intake forms and approval gates, but no AI analysis. The architecture makes dropping in AI modules for them straightforward post-MVP.
Added: ChromaDB as the embedded vector store (no separate server process, trivially swappable for Pinecone or Weaviate later). Document chunking, embedding, and retrieval as the foundation for all AI analysis.
Cut: User authentication (designed for via middleware stubs), production deployment (designed for via Docker Compose).
Workflow state machine (updated):
INTAKE
→ USE_CASE_REVIEW (Stage 1 — human form)
→ USE_CASE_APPROVED
→ LEGAL_REVIEW (Stage 2 — AI + RAG)
→ LEGAL_APPROVED
→ NDA_PENDING
→ SECURITY_REVIEW (Stage 3 — AI + RAG)
→ SECURITY_APPROVED
→ FINANCIAL_REVIEW (Stage 4 — human form)
→ FINANCIAL_APPROVED
→ ONBOARDED
/ REJECTED (exit from any stage)
vendor-onboarding/
├── backend/
│ ├── main.py
│ ├── api/routes/
│ │ ├── vendors.py
│ │ ├── documents.py
│ │ ├── reviews.py
│ │ └── decisions.py
│ ├── core/
│ │ ├── config.py
│ │ ├── database.py
│ │ └── models.py
│ ├── services/
│ │ ├── llm/
│ │ │ └── client.py # litellm wrapper
│ │ ├── document/
│ │ │ ├── extractor.py # PDF/DOCX/text → raw text
│ │ │ └── chunker.py # text → chunks
│ │ ├── rag/
│ │ │ ├── embedder.py # text → vectors
│ │ │ ├── store.py # ChromaDB wrapper
│ │ │ └── retriever.py # query → relevant chunks
│ │ ├── knowledge_base/
│ │ │ ├── legal_kb.py # regulatory requirements
│ │ │ ├── security_kb.py # control framework
│ │ │ └── loader.py # seeds KB into ChromaDB on startup
│ │ ├── legal/
│ │ │ └── analyzer.py # Stage 2 AI module
│ │ ├── security/
│ │ │ └── analyzer.py # Stage 3 AI module
│ │ └── workflow.py
│ └── requirements.txt
├── frontend/
│ └── (Vite + React + TypeScript)
├── .env.example
├── docker-compose.yml
└── README.md
Same goal as before — a running skeleton where every moving part exists. The key addition is modeling all four stages and thinking clearly about the two types of review: AI-driven (Stages 2 and 3) and human-form (Stages 1 and 4).
Data models:
Vendor — id, name, website, description, status (state machine enum), created_at.
Document — id, vendor_id, stage (enum: USE_CASE, LEGAL, SECURITY, FINANCIAL), doc_type (string: e.g. "SOC2", "privacy_policy", "use_case_brief"), filename, raw_text, chroma_collection_id, uploaded_at. The chroma_collection_id links the document to its vector store collection for retrieval later.
Review — id, vendor_id, stage, review_type (enum: AI_ANALYSIS, HUMAN_FORM), status (PENDING, COMPLETE, ERROR), ai_output (JSON, nullable — null for human-form stages), form_input (JSON, nullable — captures human form data for Stages 1 and 4), triggered_at, completed_at.
Decision — id, review_id, actor, action (APPROVE, REJECT, APPROVE_WITH_CONDITIONS), rationale, conditions (JSON array, nullable), decided_at.
AuditLog — id, vendor_id, event_type, actor, payload (JSON), timestamp.
Stage 1 form schema (define now, used in Day 5 and 6):
class UseCaseFormInput(BaseModel):
use_case_description: str
business_justification: str
data_types_involved: List[str]
estimated_users: int
alternatives_considered: str
reviewer_name: str
recommendation: Literal["PROCEED", "DO_NOT_PROCEED"]
notes: Optional[str]Stage 4 form schema:
class FinancialRiskFormInput(BaseModel):
vendor_annual_revenue: Optional[str]
years_in_operation: Optional[int]
financial_documents_reviewed: List[str]
concentration_risk_flag: bool
financial_stability_assessment: Literal["STABLE", "CONCERN", "HIGH_RISK"]
contract_value: Optional[str]
reviewer_name: str
recommendation: Literal["ACCEPTABLE", "ACCEPTABLE_WITH_CONDITIONS", "UNACCEPTABLE"]
conditions: Optional[List[str]]
notes: Optional[str]These schemas are stored in the Review.form_input JSON column when submitted. They're also used to generate the frontend forms automatically — define them once, use them everywhere.
End of Day 1 check: uvicorn main:app runs, /docs shows all route stubs, SQLite tables are created on startup, Pydantic schemas validate correctly.
This is the most foundational day and also the most important to get right. The RAG layer is what separates this from a basic "paste text into prompt" tool — it makes the system extensible to large documents, supports evidence retrieval from the knowledge base, and is the primary architectural differentiator worth building even in the MVP.
Document ingestion (services/document/):
extractor.py — same as before: PDF via pdfplumber, DOCX via python-docx, text paste passthrough. All three paths produce normalized raw_text.
chunker.py — split raw_text into overlapping chunks for embedding. Use a simple recursive character splitter (LangChain's RecursiveCharacterTextSplitter is fine here even if you don't use the rest of LangChain). Target chunk size of 512 tokens with 64-token overlap. Overlap ensures that context spanning a chunk boundary isn't lost during retrieval. Each chunk is stored with metadata: doc_id, vendor_id, stage, chunk_index, page_number (where available).
from langchain.text_splitter import RecursiveCharacterTextSplitter
class DocumentChunker:
def __init__(self, chunk_size: int = 512, overlap: int = 64):
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=overlap,
length_function=len,
)
def chunk(self, text: str, metadata: dict) -> List[Chunk]:
texts = self.splitter.split_text(text)
return [
Chunk(text=t, metadata={**metadata, "chunk_index": i})
for i, t in enumerate(texts)
]LLM client (services/llm/client.py):
Use litellm exactly as described in the previous plan. One addition: add a complete_with_json_output method that handles the parse-retry logic centrally so every downstream module gets it for free.
async def complete_with_json_output(self, system: str, user: str) -> dict:
raw = await self.complete(system, user)
try:
return json.loads(raw)
except json.JSONDecodeError:
# Strip markdown fences and retry once
cleaned = re.sub(r"```(?:json)?|```", "", raw).strip()
return json.loads(cleaned)RAG layer (services/rag/):
embedder.py — wraps an embedding model. Use sentence-transformers with all-MiniLM-L6-v2 as the default (fast, local, no API cost). Make the embedding model configurable so it can be swapped for OpenAI text-embedding-3-small or Anthropic's embeddings via env var.
class Embedder:
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
self.model = SentenceTransformer(model_name)
def embed(self, texts: List[str]) -> List[List[float]]:
return self.model.encode(texts, show_progress_bar=False).tolist()store.py — wraps ChromaDB. Use chromadb in embedded mode (no server, data persisted to a local directory). Each vendor document gets its own ChromaDB collection named vendor_{vendor_id}_{stage}_{doc_id}. The knowledge base (regulatory requirements, security controls) gets its own persistent collections named kb_legal and kb_security, seeded once on startup.
import chromadb
class VectorStore:
def __init__(self, persist_dir: str = "./chroma_data"):
self.client = chromadb.PersistentClient(path=persist_dir)
def upsert_chunks(self, collection_name: str, chunks: List[Chunk]) -> None:
collection = self.client.get_or_create_collection(collection_name)
collection.upsert(
ids=[f"{c.metadata['doc_id']}_{c.metadata['chunk_index']}" for c in chunks],
documents=[c.text for c in chunks],
metadatas=[c.metadata for c in chunks],
)
def query(self, collection_name: str, query_text: str, n_results: int = 5) -> List[str]:
collection = self.client.get_collection(collection_name)
results = collection.query(query_texts=[query_text], n_results=n_results)
return results["documents"][0]retriever.py — the interface that analyzers use. Takes a query string and a collection name, returns the top-k most relevant chunks as a single formatted string ready for prompt injection.
class Retriever:
def __init__(self, store: VectorStore):
self.store = store
def retrieve(self, query: str, collection: str, n: int = 5) -> str:
chunks = self.store.query(collection, query, n_results=n)
return "\n\n---\n\n".join(chunks)Knowledge base seeding (services/knowledge_base/loader.py):
On application startup (in main.py lifespan handler), check whether kb_legal and kb_security collections already exist in ChromaDB. If not, chunk and embed all knowledge base entries and upsert them. This is a one-time operation that takes a few seconds the first time and is a no-op thereafter.
@asynccontextmanager
async def lifespan(app: FastAPI):
await KnowledgeBaseLoader().seed_if_empty()
yieldEnd of Day 2 check: Upload a PDF, confirm chunks are stored in ChromaDB with correct metadata. Query the collection with a plain-English question ("does this vendor have encryption at rest?") and confirm relevant chunks are returned. Confirm both kb_legal and kb_security are seeded and queryable.
The analyzer now uses RAG in two ways: retrieving relevant regulatory requirements from the knowledge base, and retrieving relevant passages from the vendor document. This is the key architectural difference from the previous plan — the model is grounded in both the regulatory framework and the specific vendor evidence simultaneously.
Analysis flow:
For each regulatory requirement in the knowledge base, the analyzer issues a targeted retrieval query against the vendor document collection, fetching the most relevant chunks. It then builds a consolidated prompt that includes the requirement descriptions (retrieved from kb_legal) and the most relevant vendor document passages. This keeps prompts focused and avoids overwhelming the context window with irrelevant document sections.
class LegalAnalyzer:
async def analyze(self, vendor_id: int, doc_id: int) -> LegalAnalysisResult:
# 1. Retrieve all legal requirements from KB
requirements = self.retriever.retrieve(
query="data privacy regulatory requirements",
collection="kb_legal",
n=20
)
# 2. Retrieve relevant vendor document sections per topic
vendor_collection = f"vendor_{vendor_id}_legal_{doc_id}"
data_processing = self.retriever.retrieve("data processing agreements", vendor_collection)
data_transfers = self.retriever.retrieve("international data transfers", vendor_collection)
breach_notification = self.retriever.retrieve("breach notification incident response", vendor_collection)
retention_deletion = self.retriever.retrieve("data retention deletion", vendor_collection)
# 3. Build prompt with retrieved context
vendor_context = f"""
DATA PROCESSING:
{data_processing}
INTERNATIONAL TRANSFERS:
{data_transfers}
BREACH NOTIFICATION:
{breach_notification}
RETENTION AND DELETION:
{retention_deletion}
"""
# 4. Call LLM with structured output schema
result = await self.llm.complete_with_json_output(
system=LEGAL_SYSTEM_PROMPT,
user=LEGAL_USER_PROMPT.format(
requirements=requirements,
vendor_context=vendor_context
)
)
return LegalAnalysisResult(**result)The output schema is identical to what was defined in the original plan. The difference is that every finding's evidence field is now grounded in an actually retrieved passage rather than something the model may have hallucinated from its training data.
End of Day 3 check: Run a full legal analysis on a privacy policy (real or mock). Confirm findings cite specific retrieved passages. Test with a document that is clearly missing GDPR Article 28 language — confirm the module returns NOT_MET for that requirement with a meaningful gap description.
Same RAG pattern as Stage 2. The retrieval queries are domain-specific: one query per security control domain against the vendor's security documentation collection.
# Retrieval queries per control domain
SECURITY_RETRIEVAL_QUERIES = {
"access_control": "MFA multi-factor authentication least privilege access management",
"data_protection": "encryption at rest in transit key management data security",
"incident_response": "incident response breach notification SLA detection",
"vulnerability_management": "penetration testing patching vulnerability scanning CVE",
"business_continuity": "disaster recovery RTO RPO backup business continuity",
"supply_chain": "third party vendor assessment software composition supply chain",
}For each domain, retrieve the top 4 chunks from the vendor document and the corresponding control requirements from kb_security. Pass both into the structured prompt.
The output schema (risk score, domain assessments, critical gaps, recommendation) is identical to the original plan.
NDA gate: Implemented as a POST /vendors/{id}/confirm-nda endpoint that advances status from LEGAL_APPROVED to SECURITY_REVIEW (and simultaneously creates the Financial Review's NDA-pending state). The security analysis endpoint returns 403 with a clear error message if the vendor's status isn't SECURITY_REVIEW or later.
End of Day 4 check: Run analysis on a SOC 2 report or mock security questionnaire. Confirm domain-level risk scores are populated. Confirm NDA gate blocks analysis when status is LEGAL_APPROVED and allows it after confirm-nda is called.
Today you wire all four stages into a single orchestrated workflow, implement the Stage 1 and Stage 4 placeholder form submission endpoints, and build the audit log.
Stage 1 — Use Case Evaluation (placeholder):
The intake endpoint creates a vendor record and a Stage 1 review in PENDING state. The Stage 1 review is completed not by AI analysis but by a human submitting the UseCaseFormInput schema via POST /reviews/{id}/submit-form. The workflow service validates the form, stores it in Review.form_input, marks the review COMPLETE, and if recommendation == "PROCEED", advances vendor status to USE_CASE_APPROVED and automatically creates a Legal Review in PENDING state (ready to be triggered when documents are uploaded).
This means Stage 1 is a real gate — a vendor cannot reach Stage 2 without a human submitting the use case form with a PROCEED recommendation. The form data is stored, displayed in the UI, and included in the audit trail. It's just not AI-analyzed.
Stage 4 — Financial Risk Evaluation (placeholder):
Identical pattern. After Security is approved, vendor status advances to FINANCIAL_REVIEW. A human submits the FinancialRiskFormInput schema. On PROCEED/ACCEPTABLE, vendor status advances to FINANCIAL_APPROVED. If all four stages are approved, a final POST /vendors/{id}/complete-onboarding call sets status to ONBOARDED.
Workflow service (all four stages):
class WorkflowService:
# Stage 1
def create_vendor_and_intake(self, data: VendorCreate) -> Vendor: ...
def submit_use_case_form(self, review_id: int, form: UseCaseFormInput) -> Decision: ...
# Stage 2
def trigger_legal_review(self, vendor_id: int, doc_id: int) -> Review: ...
def submit_legal_decision(self, review_id: int, action: str, rationale: str) -> Decision: ...
# NDA
def confirm_nda(self, vendor_id: int) -> Vendor: ...
# Stage 3
def trigger_security_review(self, vendor_id: int, doc_id: int) -> Review: ...
def submit_security_decision(self, review_id: int, action: str, rationale: str) -> Decision: ...
# Stage 4
def submit_financial_form(self, review_id: int, form: FinancialRiskFormInput) -> Decision: ...
# Final
def complete_onboarding(self, vendor_id: int) -> Vendor: ...
def reject_vendor(self, vendor_id: int, stage: str, rationale: str) -> Vendor: ...Every method appends to the audit log before returning. Define a private _log(vendor_id, event_type, actor, payload) helper used by all methods.
Auth stub (designed for, not implemented):
Add a get_current_user dependency in core/auth.py that currently returns a hardcoded {"id": "dev_user", "name": "Dev User", "role": "admin"}. Every route that takes a decision already accepts this dependency. Swapping in real JWT/session auth post-MVP requires only implementing this one function.
End of Day 5 check: Walk the complete four-stage lifecycle via the API: create vendor + submit use case form → upload legal doc + trigger legal review → approve → confirm NDA → upload security doc + trigger security review → approve → submit financial form → complete onboarding. Verify every step appears in the audit log.
The frontend now covers all four stages. Stages 1 and 4 render dynamic forms; Stages 2 and 3 render AI analysis reports. The structure is the same as before with additions.
Stage 1 — Use Case Form:
Render the UseCaseFormInput schema as a form. Fields: text area for use case description, text area for business justification, multi-select for data types (Personal Data, Financial Data, Health Data, Employee Data, Public Data), number input for estimated users, text area for alternatives considered, text input for reviewer name, radio for recommendation (Proceed / Do Not Proceed), optional notes. On submit, POST /reviews/{id}/submit-form.
Stage 4 — Financial Risk Form:
Render the FinancialRiskFormInput schema. Fields: optional text inputs for revenue and years in operation, checkbox list for documents reviewed (Financial Statements, Credit Report, Insurance Certificates, etc.), checkbox for concentration risk flag, radio for stability assessment, optional contract value, reviewer name, radio for recommendation with optional conditions list. On submit, POST /reviews/{id}/submit-form.
Stage 2 and 3 — AI Reports:
Same as described in the original plan. Compliance matrix table for Stage 2, domain risk cards for Stage 3. RAG-grounded evidence is displayed in the findings table — users can see exactly which passage from the vendor document triggered each finding.
Vendor Detail page — updated stepper:
The status stepper now shows all four stages: Use Case → Legal Review → (NDA) → Security Review → Financial Review → Decision. Each step shows its status (Not Started / In Progress / Approved / Rejected). The NDA confirmation button appears as a prompt between Legal and Security when status is LEGAL_APPROVED.
Post-MVP extension points to make visible in UI:
Add a disabled "Authentication" nav item with a tooltip "Coming soon — add user auth to enable." Add a disabled "Export Report" button on the vendor detail page. These are intentional signals that the architecture has room to grow, which matters for a demo or portfolio context.
End of Day 6 check: Complete the full four-stage workflow through the UI. Stage 1 and 4 forms submit correctly and advance the workflow. Stage 2 and 3 AI results render with evidence citations. Audit trail page shows all events in order.
Mock vendor scenarios (three):
Vendor A — "Acme Analytics" (clean pass): A straightforward SaaS analytics vendor with a solid privacy policy covering GDPR and PIPEDA, a SOC 2 Type II report with no exceptions, and clean financials. Should produce PROCEED on all AI stages and give the reviewer an easy approval flow.
Vendor B — "DataFlow Inc." (legal issues): A vendor whose privacy policy mentions data processing but is silent on international transfer mechanisms and has no GDPR Article 28 processor agreement language. Should produce a DOES_NOT_MEET on Stage 2 with clear gap citations, allowing you to demonstrate the rejection path.
Vendor C — "SecureVault Ltd." (security gaps, financial conditions): Passes legal review cleanly but the security documentation is vague on incident response SLA and has no evidence of third-party penetration testing. Stage 3 returns ACCEPTABLE_WITH_CONDITIONS. Stage 4 human reviewer flags a concentration risk concern and marks ACCEPTABLE_WITH_CONDITIONS. Demonstrates the conditional path.
Write these as .txt files in backend/mock_data/ — realistic but concise (300–600 words each). They should be loadable via a POST /dev/seed endpoint that creates all three vendors and uploads their documents in one call, so you can reset the demo state instantly.
RAG quality check:
Run the three mock scenarios and manually verify that the retrieved chunks in each finding are genuinely relevant to the finding. If a finding about "encryption at rest" is citing a passage about "data retention schedules," the retrieval query needs tuning. Adjust the SECURITY_RETRIEVAL_QUERIES dict until evidence citations are meaningfully grounded.
README:
git clone ...
cd vendor-onboarding
cp .env.example .env # add your API key
docker compose up # or: cd backend && pip install -r requirements.txt && uvicorn main:app
# Frontend: cd frontend && npm install && npm run dev
# Seed demo data: POST http://localhost:8000/dev/seed
Five commands, running app, demo data loaded.
Demo script (3 minutes):
Start with Vendor A to show the happy path — submit use case form, upload privacy policy, watch AI legal analysis load and show all green, approve, confirm NDA, upload SOC 2, watch security analysis load, show domain cards, approve, submit financial form, vendor is ONBOARDED. Then switch to Vendor B and show Stage 2 rejecting with specific gap citations. Show the audit trail for Vendor B. That's the whole demo.
The RAG layer adds complexity primarily on Day 2. The risk is embedding quality — all-MiniLM-L6-v2 is fast and free but not as semantically precise as OpenAI embeddings. If retrieval quality is poor during Day 3 testing, swap the embedder to text-embedding-3-small (add OPENAI_API_KEY to .env and change one env var). Keep this swap in mind as a Day 3 contingency rather than a surprise.
The four-stage workflow adds surface area to the frontend on Day 6. If time is tight, the Stage 4 financial form is the lowest-priority UI element — it can be temporarily replaced with a simple approve/reject button without breaking the demo narrative.