The Fine Print

import os
from typing import TypedDict, List
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
class VendorException(BaseModel):
    control_id: str = Field(description="Control ID, e.g., CC6.1")
    description: str = Field(description="Description of exception")
    has_compensating_control: bool = Field(description="True if compensating control identified")
    risk_level: str = Field(description="HIGH, MEDIUM, or LOW")
class VendorRiskState(TypedDict):
    report_text: str
    parsed_exceptions: List[dict]
    overall_risk_score: str
    requires_human_review: bool
    final_assessment: str
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def parser_node(state: VendorRiskState) -> dict:
    structured_llm = llm.with_structured_output(VendorException)
    prompt = f"Analyze the following SOC2 report text for exceptions and controls:\n{state['report_text']}"
    res = structured_llm.invoke(prompt)
    exceptions = [res.model_dump()]
    return {"parsed_exceptions": exceptions}
def risk_evaluator_node(state: VendorRiskState) -> dict:
    exceptions = state["parsed_exceptions"]
    high_risks = [e for e in exceptions if e.get("risk_level") == "HIGH" and not e.get("has_compensating_control")]
    if high_risks:
        return {"overall_risk_score": "HIGH", "requires_human_review": True}
    return {"overall_risk_score": "LOW", "requires_human_review": False}
def generate_assessment_node(state: VendorRiskState) -> dict:
    score = state["overall_risk_score"]
    review = state["requires_human_review"]
    summary = f"Vendor Risk Rating: {score}. Requires Escalation: {review}. Total Exceptions Evaluated: {len(state['parsed_exceptions'])}"
    return {"final_assessment": summary}
def route_risk(state: VendorRiskState) -> str:
    if state["requires_human_review"]:
        return "escalate"
    return "approve"
def escalation_node(state: VendorRiskState) -> dict:
    assessment = state["final_assessment"] + " | ESCALATED: Requires CISO Sign-off due to unmitigated high risk."
    return {"final_assessment": assessment}
workflow = StateGraph(VendorRiskState)
workflow.add_node("parse", parser_node)
workflow.add_node("evaluate", risk_evaluator_node)
workflow.add_node("assess", generate_assessment_node)
workflow.add_node("escalate", escalation_node)
workflow.set_entry_point("parse")
workflow.add_edge("parse", "evaluate")
workflow.add_edge("evaluate", "assess")
workflow.add_conditional_edges("assess", route_risk, {"escalate": "escalate", "approve": END})
workflow.add_edge("escalate", END)
app = workflow.compile()
if __name__ == "__main__":
    test_report = "Control CC7.2 Exception: Production database access was granted without ticket approval in 3 instances. No compensating automated audit logs were available."
    res = app.invoke({"report_text": test_report, "parsed_exceptions": [], "overall_risk_score": "", "requires_human_review": False, "final_assessment": ""})
    print(res["final_assessment"])