Governance, Risk, and Compliance (GRC) with LangGraph
Problem Statement
NIST CSF 2.0 expanded its core functions by introducing the GOVERN (GV) function. Organizations require continuous scoring of maturity levels across NIST CSF categories. Static spreadsheets result in stale data. An agentic graph can continuously collect operational security metrics, map them to CSF subcategories, and recalculate maturity scores dynamically.
Goal
Build a LangGraph state machine that evaluates security metrics against NIST CSF 2.0 subcategories (e.g., GV.RM-01, PR.AA-01) and generates quantitative maturity scores (Tiers 1-4).
Outcome Measurement
- Assessment Frequency: Continuous real-time scoring vs. once-a-year annual spreadsheets.
- Audit Readiness: Audit evidence mapping time reduced by 90%.
How the Code was Built
- Uses Pydantic structured output models to force strictly formatted score objects (Tier 1-4).
- Aggregates multi-source evidence into categorical NIST CSF subcategories via graph nodes.
Executable Code
from typing import TypedDict, Listfrom pydantic import BaseModel, Fieldfrom langgraph.graph import StateGraph, ENDfrom langchain_openai import ChatOpenAIclass CSFEvaluation(BaseModel): subcategory: str = Field(description="NIST CSF Subcategory ID, e.g., GV.RM-01") tier_score: int = Field(description="Maturity Tier from 1 (Partial) to 4 (Adaptive)") justification: str = Field(description="Audit justification for assigned tier")class CSFState(TypedDict): subcategory: str evidence_inputs: List[str] evaluation: dictllm = ChatOpenAI(model="gpt-4o", temperature=0)def evaluate_csf_subcategory(state: CSFState) -> dict: structured_llm = llm.with_structured_output(CSFEvaluation) prompt = f"""Evaluate NIST CSF 2.0 Subcategory: {state['subcategory']}Evidence Provided:{chr(10).join(state['evidence_inputs'])}Assign a maturity Tier (1 to 4) and provide justification.""" res = structured_llm.invoke(prompt) return {"evaluation": res.model_dump()}workflow = StateGraph(CSFState)workflow.add_node("evaluate", evaluate_csf_subcategory)workflow.set_entry_point("evaluate")workflow.add_edge("evaluate", END)app = workflow.compile()if __name__ == "__main__": data = { "subcategory": "GV.RM-01: Risk management objectives are established and agreed to by organizational stakeholders.", "evidence_inputs": [ "Executive board approved Risk Tolerance Statement dated Jan 2026.", "Quarterly risk reviews conducted with documented sign-offs." ], "evaluation": {} } out = app.invoke(data) print("NIST CSF Evaluation Result:") print(out["evaluation"])






Leave a Reply