Griptape Governance Pipeline

“Disclaimer: This article documents my personal exploration of AI governance frameworks and regulatory concepts. Nothing in this post constitutes legal advice, compliance certification, or a guarantee of regulatory conformance. AI governance requirements vary by jurisdiction, sector, and use case. Consult qualified legal and compliance professionals before making governance decisions. NosisTech LLC accepts no liability for outcomes arising from use of this content.”

I rebuilt the Griptape architecture as a small governance pipeline because the important lesson was not the framework itself. The lesson was discipline. A useful enterprise agent needs more than a model call. It needs a controlled path from request, to policy context, to model judgment, to a human checkpoint when the risk profile calls for one.


What This Agent Does

This agent takes three fictional enterprise requests and routes each one through a fixed pipeline. A pipeline is a sequence of tasks that run in order, which makes the agent easier to inspect than a free-form chat loop.

The agent validates the request, matches local policy context, stores supporting context in temporary task memory, asks the model for a JSON decision, and prints a short decision packet. I tested the same code through LiteLLM on gemini-flash and deepseek-v4-pro.


Why I Built This

What stood out to me in Griptape was not just agents, tools, and memory. It was the idea that serious AI systems need structure around the model.

In governance work the value comes from knowing which step happened, which context was used, and where human review entered the process.

I wanted the smallest version of that pattern I could run and test. The result is not a Griptape clone. It is a LiteLLM-native rebuild of the architectural idea: a controlled pipeline that makes the model one step in a larger decision process.


What Actually Happened

What broke: The first runtime test exposed provider differences. Gemini wrapped or formatted JSON in a way my parser rejected, and DeepSeek returned the checkpoint flag in a type the validator did not accept.

What surprised me: The architecture worked on both providers after a small parsing fix. The same pipeline produced the intended low, medium, and high risk behavior without provider-specific code.

What I would change next: I would add a tiny audit log around each task. That would turn this from a governance pipeline into a stronger evidence trail.

Why this matters: The build showed that agent governance is not only about the model answer. It is about the steps around the model that make the answer inspectable.


The Architecture: How It Works

The agent uses four ideas.

First, it uses a fixed pipeline. Each request passes through the same task sequence every time.

Second, it uses local policy matching. The model does not receive every possible policy. The code selects the relevant context first.

Third, it uses task memory. The full policy context is stored under a memory reference, while the prompt receives only the selected excerpt and reference key.

Fourth, it validates the model response as JSON. JSON is a structured data format, which means the code can check whether the model returned the fields the program expects.


The Build: Step by Step

POLICIES = [
    (["export", "customer", "financial"], "Ardent Harbor Finance: customer financial exports require steward review and CISO sign-off."),
    (["patient", "medical", "phi"], "Luma Ridge Health: patient health information requires Privacy Officer approval."),
    (["vendor", "supplier", "third-party", "access"], "Northstar Parts: vendor system access requires manager and IT Security approval."),
]

SCENARIOS = [
    ("Safe summary", "Summarize internal team productivity notes for engineering leads."),
    ("Vendor access", "Grant a third-party supplier temporary access to the parts ordering system."),
    ("Customer export", "Export customer financial records from the past five years."),
]

What This Code Is Actually Doing

This is the test bench. The policies are fictional rules that give the agent something to reason against. The scenarios create three different outcomes: one low-risk summary, one vendor access request, and one customer data export request.

def retrieve_policy_context(state):
    """Match local policy context and store the full text off-prompt."""
    request_lower = state["request"].lower()
    matches = [text for keys, text in POLICIES if any(key in request_lower for key in keys)]
    context = "\n".join(matches) or "No matching policy found."
    memory_key = "mem-" + str(len(state["memory"]) + 1)
    state["memory"][memory_key] = context
    state.update(policy_excerpt=context.split("\n")[0], policy_refs=[memory_key])
    state["risk"] = score_risk(state["request"], bool(matches))
    return state

What This Code Is Actually Doing

This function acts like a small retrieval layer. It checks the request for policy keywords, saves the matching policy text in temporary memory, and passes only a policy excerpt plus a memory reference forward. That is the off-prompt memory pattern: the system keeps supporting context available without stuffing everything into the model prompt.

def score_risk(request, matched_policy):
    """Score risk with simple transparent rules."""
    request_lower = request.lower()
    if matched_policy and any(word in request_lower for word in ["export", "patient", "financial", "phi"]):
        return "HIGH"
    if matched_policy:
        return "MEDIUM"
    return "LOW"

What This Code Is Actually Doing

This is the local risk gate. The model is not asked to invent the risk level from scratch. The code creates a simple risk signal first, then the model reviews the request with that signal already attached.

def assess_request_with_model(state):
    """Ask the LiteLLM-routed model for a structured decision."""
    prompt = (
        "Return one JSON object only. Keys: decision, reason, required_human_checkpoint. "
        "decision must be APPROVE, REVIEW, or BLOCK. required_human_checkpoint must be a boolean.\n\n"
        "Request: " + state["request"] + "\nPolicy excerpt: " + state["policy_excerpt"] + "\nLocal risk: " + state["risk"] + "\n"
        "Memory reference: " + ", ".join(state["policy_refs"])
    )
    decision = call_model(state["client"], prompt)
    state["decision"] = clean_decision(decision, state["risk"])
    return state

What This Code Is Actually Doing

This is where the model enters the pipeline. The prompt gives the model the request, the policy excerpt, the local risk level, and the memory reference. The model is asked for a JSON decision so the code can inspect the result instead of treating the answer as plain text.

def parse_json_object(raw_content):
    """Parse a JSON object even when a provider wraps it in extra text."""
    content = raw_content.strip()
    if content.startswith("```"):
        content = content.replace("```json", "").replace("```", "").strip()
    start_index = content.find("{")
    end_index = content.rfind("}")
    if start_index == -1 or end_index == -1:
        raise json.JSONDecodeError("No JSON object found.", content, 0)
    return json.loads(content[start_index : end_index + 1])

What This Code Is Actually Doing

This function handles a real provider difference I hit during testing. Gemini returned output that the first parser did not accept cleanly. Instead of creating provider-specific branches, I added a small parser that extracts the JSON object from the response.

def clean_decision(decision, risk):
    """Validate model JSON and enforce human review for higher risk."""
    if not isinstance(decision, dict):
        return review_decision("The model response was not a JSON object.")
    if decision.get("required_human_checkpoint") in {"true", "True"}:
        decision["required_human_checkpoint"] = True
    if decision.get("required_human_checkpoint") in {"false", "False"}:
        decision["required_human_checkpoint"] = False
    if (
        decision.get("decision") not in {"APPROVE", "REVIEW", "BLOCK"}
        or not isinstance(decision.get("reason"), str)
        or not isinstance(decision.get("required_human_checkpoint"), bool)
    ):
        return review_decision("The model response did not match the decision schema.")
    if risk in {"MEDIUM", "HIGH"}:
        decision["required_human_checkpoint"] = True
    return decision

What This Code Is Actually Doing

This is the contract check. A contract is the shape of data the program expects. If the model returns an unsupported decision, missing reason, or bad checkpoint flag, the system routes the case to review instead of trusting the response.

def run_pipeline(client, scenario, request):
    """Run the fixed task pipeline for one request."""
    state = {"client": client, "scenario": scenario, "request": request, "memory": {}}
    pipeline = [validate_request, retrieve_policy_context, assess_request_with_model, print_decision_packet]
    for task in pipeline:
        state = task(state)

What This Code Is Actually Doing

This is the Griptape-style idea in its smallest form. The request moves through a fixed list of tasks. Each task receives the current state, changes it, and passes it to the next task.


Errors I Hit During This Build

The first issue was size. The first model-generated version of agent.py worked as a draft, but it was much larger than needed. I cut it down to a small pipeline that keeps the architecture visible.

The second issue appeared with DeepSeek. The output showed: “The model did not return a valid checkpoint flag.” The cause was a provider response where the checkpoint value did not arrive as the exact boolean type my validator expected. I fixed it by normalizing string values such as true and false before the final schema check.

The third issue appeared with Gemini. The output showed: “The model did not return valid JSON.” The cause was response formatting around the JSON object. I fixed it by extracting the JSON object from the response before parsing.

The final test passed on both providers. gemini-flash and deepseek-v4-pro both approved the safe summary request and routed the vendor access and customer export requests to review with a human checkpoint.


Why This Matters

This build matters because governance is a workflow problem before it is a model problem.

The model can help judge a request, but the system around the model decides what context the model sees, how the response is checked, and when a human checkpoint is attached. That is where the real architecture lives.

For enterprise AI, this pattern has leverage. A disciplined pipeline can become an audit trail, a billing ledger, a tenant isolation layer, or a compliance review gate. The small version I built here is the seed of those larger systems.


Tools Used

Griptape (Apache 2.0): github.com/griptape-ai/griptape
LiteLLM (MIT): github.com/BerriAI/litellm
OpenAI Python SDK (Apache 2.0): github.com/openai/openai-python
python-dotenv (BSD-3-Clause): github.com/theskumar/python-dotenv
NosisTech Agent Engineering (MIT): github.com/nosistech/agent-engineering


(c) 2026 NosisTech LLC. Licensed under CC BY 4.0. Use freely, just credit us.

WEEKLY BUILD NOTES

One documented agent build in your inbox, every week

Real systems, real code, the errors left in. No spam, unsubscribe anytime.