Engineer Reliable AI Solutions
Add deterministic AI evals and CI gates to a support triage service.
Introduction
30 Second Summary
A support team can look fast until one urgent customer issue reaches the wrong people. One missed security report can turn a tidy queue into a client crisis.
In this project, you will turn your Part 1 FastAPI service into a guarded support issue triage service. A two-tier regression gate checks deterministic rules plus Gemini-assisted behavior before changes merge.
What You'll Build
Picture a support issue flowing through POST /triage to produce a guarded decision that survives all 20 regression checks.
By the end of this project, you'll have:
- A live triage endpoint that returns priority, routing team, review status, decision source, plus a model-assisted summary.
- A two-tier regression suite covering 10 direct rule checks plus 10 stubbed orchestrator requests without a live Gemini call.
- A merge-blocking gate in GitHub Actions that runs the architecture policy beside both evaluation tiers.
- Secret Mission: Trigger a forbidden SDK import failure. Repair the provider boundary without changing service behavior.
Are there any prerequisites?
This project assumes your completed Part 1 repository is available on Linux with its original Gemini key stored in .env.
The four Part 2 starter assets are supplied, so you do not need another credential or service.
Before We Start
Before any hands-on work, lock in why your production evaluation gate matters. Identify the support-triage failure that would cause the most damage for a client, then explain why the gate must catch it before merge.
Set Up the Evaluation Extension
Your Part 1 FastAPI service already has the foundation for a production evaluation gate. Rebuilding that environment would distract from the rules you need to test.
This step keeps the existing credential in place. You will isolate Part 2 on a Git branch. You will also prepare the supplied assets inside the existing repository.
In this step, get ready to:
- Create a Part 2 branch without exposing the existing Gemini key.
- Map the supplied assets to their repository locations.
- Activate a Python 3.11 environment with every pinned dependency installed.
Protect the Part 1 baseline
A separate branch keeps the completed Part 1 version available while you add the evaluation layer. The existing .env stays local throughout the change.
- Return to the Linux terminal from Part 1.
- Read the terminal prompt to confirm you are inside the existing Part 1 repository.
- Create the Part 2 branch by running this command:
git switch -c part-2-evaluation-gateWhat does this command do?
The -c option creates the branch from your current commit. Git switches your working tree to that branch immediately.
Good start. Your evaluation work now has its own branch while the Part 1 baseline remains intact.
Checking a credential file can feel risky. This next command prints only the variable name plus its equals sign, so the key value remains hidden.
- Confirm the expected key name exists by running this safe check:
grep -o '^GEMINI_API_KEY=' .envHow does this protect the key?
The -o option prints only the text matched by the pattern. The pattern ends at the equals sign before the credential value begins.
You should see GEMINI_API_KEY= on its own. Your terminal should show no part of the secret value.
Branch or key check failed?
If Git cannot create the branch, check whether that branch name already exists. Switch to the existing Part 2 branch before continuing.
If the key name does not appear, confirm that .env is at the top level of the existing repository.
Help me diagnose the branch or safe key check.
Organize the evaluation assets
The supplied files use repository paths to show where each responsibility belongs. This separation keeps service code away from evaluation policy and delivery automation.
Where does each asset belong?
- Service files map to app/.
- Fixture data plus evaluation tests map to evals/.
- Import-policy code maps to architecture/.
- The C4 document maps to docs/.
- The endpoint caller maps to scripts/.
- The automation workflow maps to .github/workflows/.
- In the Part 1 repository workspace, set the top-level requirements.txt file to the supplied dependency manifest below:
fastapi==0.147.0
pydantic==2.14.1
uvicorn[standard]==0.54.0
google-genai==2.31.0Why pin every dependency?
Each double equals sign locks one package to a tested version. The local environment and the later CI gate can now use the same dependency set.
- Save requirements.txt.
- Confirm requirements.txt appears at the top level of the repository file tree.
- In the repository file tree, create app/__init__.py with this module description:
"""Support issue triage service."""What does this file establish?
The module docstring identifies app as the support issue triage package. Python can import the service modules through that package path.
- Save app/__init__.py.
- Confirm __init__.py appears inside app/.
- In the repository file tree, create evals/__init__.py with this module description:
"""Evaluation suite for support issue triage."""Why make evals a package?
The module docstring identifies evals as the evaluation suite. Later test commands can address its modules through stable import paths.
- Save evals/__init__.py.
- Confirm __init__.py appears inside evals/.
File in the wrong location?
Check each relative path from the top level of the Part 1 repository. A nested duplicate folder changes the import path.
Help me check the three starter-file locations.
✔️ Awesome, I've got everything!
Great. Double check that all three starter files are saved before continuing.
ⓧ I'd like to double check the full code
Compare each starter file with these exact contents.
"""Support issue triage service.""""""Evaluation suite for support issue triage."""fastapi==0.147.0
pydantic==2.14.1
uvicorn[standard]==0.54.0
google-genai==2.31.0Prepare the pinned Python environment
A virtual environment isolates this repository's packages from the rest of your Linux device. The version check comes first because every pinned dependency requires a compatible Python interpreter.
- Check the Python interpreter available in the existing terminal by running this command:
python --version✔️ I see version 3.11 or higher
Your Python interpreter meets the project requirement. Continue below to create the isolated environment.
ⓧ I see an older version
The current interpreter is below the required 3.11 release. Install a compatible interpreter before creating the environment.
- Use the official Python downloads page to install Python 3.11 or higher.
- Return to the existing Linux terminal after installation.
- Confirm the updated interpreter by running this command:
python --versionContinue when the output reports Python 3.11 or higher.
ⓧ Command not found
Python is unavailable through the current terminal command. Install the required interpreter before continuing.
- Use the official Python downloads page to install Python 3.11 or higher.
- Return to the existing Linux terminal after installation.
- Confirm Python is available by running this command:
python --versionContinue when the command prints Python 3.11 or higher.
- Create or refresh the repository's .venv environment by running this command:
python -m venv .venvWhat does this command create?
The venv module prepares an isolated Python installation inside .venv. Project packages installed after activation stay inside that directory.
- Activate the Linux environment by running this command:
source .venv/bin/activateWhat changes after activation?
Activation places the environment's executable directory first on your shell path. Your prompt should now include (.venv).
- Install the pinned dependencies from requirements.txt by running:
python -m pip install -r requirements.txtWhat gets installed?
The manifest installs FastAPI, Pydantic, Uvicorn, and the Google Gen AI SDK at their pinned versions.
The first installation can take about a minute while package files download. A steady stream of package names means the installer is still working.
Dependency installation failed?
Confirm your prompt begins with (.venv). Check that requirements.txt matches the supplied manifest exactly.
Help me diagnose the dependency installation.
If the installation still fails, share the non-secret error output in the NextWork community.
Before you run the final check, do you expect the active environment to report the required Python release plus all four pinned package versions?
- Verify the interpreter and installed packages by running these commands:
python --version
python -m pip show fastapi pydantic google-genai uvicornWhat does the final check prove?
The first command identifies the Python interpreter used by the active shell. The second command reads the installed package metadata from that same environment.
You should see Python 3.11. You should also find FastAPI 0.147.0, Pydantic 2.14.1, Uvicorn 0.54.0, and google-genai 2.31.0 in the package details.
Seeing a missing or different version?
Check that the terminal prompt still includes (.venv). Rerun the dependency installation from the top level of the repository if one package is missing.
Help me compare the installed package versions.
That is the environment locked down. Next, you will use the labeled issue records to reveal where a plausible baseline fails.
Build the Baseline and Expose Its Gaps
Your Part 2 environment is ready. You can now turn the existing service into a support triage system with decisions that are measurable against client expectations.
A broad keyword matcher can produce plausible output while missing important business rules. In this step, frozen labels and a deterministic baseline reveal whether those plausible decisions are actually safe.
In this step, get ready to:
- Define the triage data contract and ten frozen issue records.
- Build an initial classifier that recognizes broad issue categories.
- Run Tier A as named subtests and inspect the mismatched decisions.
Define the models and frozen cases
The service needs one shared contract for issues and triage decisions. Pydantic models keep the rule engine and evaluation harness aligned around the same fields.
- In the file sidebar of the editor you use for the Part 1 repository, create app/models.py.
- Add the shared label types and human override model by pasting this first section:
from typing import Literal
from pydantic import BaseModel
Priority = Literal["P0", "P1", "P2", "P3"]
RoutingTeam = Literal["security", "payments", "reliability", "platform", "product-support"]
DecisionSource = Literal["baseline", "human_override", "model-assisted", "baseline_fallback"]
class HumanOverride(BaseModel):
priority: Priority
routing_team: RoutingTeam
reason: strWhat does this section define?
- The literal types limit priorities, routing teams, and decision sources to known values.
- The HumanOverride model captures a person's chosen priority, team, and reason.
- These constraints make invalid labels fail at the data boundary.
- Continue in app/models.py by adding the incoming issue model below HumanOverride:
class Issue(BaseModel):
issue_id: str
title: str
body: str
customer_impact: str = ""
human_override: HumanOverride | None = NoneWhat does the issue model hold?
The Issue model holds the support report and its customer impact. Its optional human_override field carries an explicit human decision when one exists.
- Save app/models.py.
- Check the partial model file for valid Python syntax by running this command:
python -m py_compile app/models.pyWhat should you see?
The py_compile check returns you to the prompt without an error when the file has valid syntax. This confirms the first two models can be imported by later code.
- Return to app/models.py.
- Add the model suggestion contract below Issue:
class ModelSuggestion(BaseModel):
priority: Priority
routing_team: RoutingTeam
summary: strWhat does a model suggest?
The ModelSuggestion contract limits model-assisted output to a priority, routing team, and summary. Later orchestration can compare this suggestion with the deterministic result.
- Complete app/models.py by adding the final decision model below ModelSuggestion:
class TriageDecision(BaseModel):
issue_id: str
priority: Priority
routing_team: RoutingTeam
requires_human_review: bool
decision_source: DecisionSource
summary: strWhat makes a complete decision?
The TriageDecision model records the guarded labels and whether a person must review the result. It also exposes the source and summary needed to explain the decision.
- Save app/models.py.
- Confirm the complete model file has valid syntax by running this command:
python -m py_compile app/models.pyWhat did this check prove?
A clean return to the prompt confirms all seven project identifiers are defined in valid Python. The evaluation files can now import the same contract as the classifier.
Seeing a model syntax error?
Check that each field inside a model is indented by four spaces. Confirm that HumanOverride | None appears exactly on the human_override field.
Ask for help with the failing line: help me diagnose the syntax error in my Pydantic models
The evaluation fixture acts as the client-approved answer key. Each JSON record pairs an issue with frozen priority, routing, and review labels.
- Create evals/issues.json inside the existing evals/ folder.
- Add the ten supplied issue records by pasting this fixture:
[
{"issue":{"issue_id":"ISSUE-001","title":"Production API credentials leaked","body":"Production API credentials leaked in a public log and customer data exposed.","customer_impact":"Credentials and customer data may be accessible externally."},"expected":{"priority":"P0","routing_team":"security","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-002","title":"Checkout and payment processing down for all customers","body":"Every checkout attempt is failing in production.","customer_impact":"All customers are unable to pay."},"expected":{"priority":"P0","routing_team":"payments","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-003","title":"Possible auth bypass","body":"One user may view another profile through an auth bypass.","customer_impact":"Potential unauthorized access requires investigation."},"expected":{"priority":"P1","routing_team":"security","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-004","title":"Multiple customers cannot log in","body":"Login intermittently returns an error.","customer_impact":"Multiple customers are blocked from their accounts."},"expected":{"priority":"P1","routing_team":"reliability","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-005","title":"Webhook signature rejected after secret rotation","body":"Our webhook integration rejects valid callbacks after rotating the signing secret.","customer_impact":"One integration cannot process callbacks."},"expected":{"priority":"P2","routing_team":"platform","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-006","title":"Dashboard is slow in one region","body":"Dashboard requests take eight seconds for users in one region.","customer_impact":"The dashboard remains usable but is slow."},"expected":{"priority":"P2","routing_team":"reliability","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-007","title":"Invoice total is incorrect","body":"The latest invoice applies a duplicate charge to one account.","customer_impact":"One customer has an incorrect bill."},"expected":{"priority":"P2","routing_team":"payments","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-008","title":"OAuth guide uses the wrong callback parameter","body":"The documentation example is wrong and causes setup confusion.","customer_impact":"Developers can work around the docs error."},"expected":{"priority":"P3","routing_team":"platform","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-009","title":"Feature request for dark mode","body":"Our support team would like dark mode in the admin console.","customer_impact":"No workflow is blocked."},"expected":{"priority":"P3","routing_team":"product-support","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-010","title":"Routine access report needs audit review","body":"The report succeeded, but an auditor requested a security review.","customer_impact":"No outage is present.","human_override":{"priority":"P1","routing_team":"security","reason":"Auditor requested immediate review."}},"expected":{"priority":"P1","routing_team":"security","requires_human_review":true}}
]Why freeze these labels?
The expected fields describe the behavior the client needs from the service. Keeping them fixed prevents the implementation from redefining success whenever a case exposes a weakness.
The fixture includes security exposure, outages, payments, integrations, performance, documentation, a feature request, and a human override. That range makes plausible but unsafe classifications visible.
- Save evals/issues.json.
- Validate the fixture and print its parsed structure by running this command:
python -m json.tool evals/issues.jsonWhat should you see?
The json.tool command prints a formatted array containing records from ISSUE-001 through ISSUE-010. Reaching the final record without a parsing error confirms the fixture is valid JSON.
Fixture validation failed?
Check for a missing comma between records. Confirm that every property name and string value uses double quotes.
Ask for help locating the malformed record: help me find the invalid JSON in my evaluation fixture
Build the broad keyword baseline
The first classifier uses broad category keywords to make deterministic decisions. Its simplicity gives the evaluation suite a concrete implementation to challenge.
- Create app/triage_rules.py inside the existing app/ folder.
- Add the phrase matcher and shared decision builder by pasting this first section:
from collections.abc import Iterable
from app.models import Issue, TriageDecision
def _contains(text: str, phrases: Iterable[str]) -> bool:
return any(phrase in text for phrase in phrases)What does the matcher do?
The _contains() helper checks whether any known phrase appears in the combined issue text. Centralizing this operation keeps each category check concise.
- Continue in app/triage_rules.py by adding the decision builder below _contains():
def _decision(issue: Issue, priority: str, team: str, reason: str) -> TriageDecision:
return TriageDecision(
issue_id=issue.issue_id,
priority=priority,
routing_team=team,
requires_human_review=priority in {"P0", "P1"} or team == "security",
decision_source="baseline",
summary=reason,
)What does the builder guarantee?
The _decision() helper returns the same response shape for every branch. It also marks high-priority or security decisions for human review.
- Save app/triage_rules.py.
- Check the helper functions for valid Python syntax by running this command:
python -m py_compile app/triage_rules.pyWhat did this check prove?
A clean return to the prompt confirms both helpers compile. The file is ready for the classifier that uses them.
- Return to app/triage_rules.py.
- Add the initial classify_issue() function below _decision():
def classify_issue(issue: Issue) -> TriageDecision:
text = " ".join([issue.title, issue.body, issue.customer_impact]).casefold()
payment = {"payment", "checkout", "invoice", "charge", "billing"}
integration = {"webhook", "api integration", "sdk", "oauth callback"}
performance = {"slow", "latency", "timeout", "performance"}
documentation = {"documentation", "docs", "guide", "typo", "example is wrong"}
feature = {"feature request", "dark mode", "enhancement", "would like"}
if _contains(text, payment):
return _decision(issue, "P2", "payments", "Payment or billing defect detected.")
if _contains(text, integration):
return _decision(issue, "P2", "platform", "Integration defect detected.")
if _contains(text, performance):
return _decision(issue, "P2", "reliability", "Performance degradation detected.")
if _contains(text, documentation):
return _decision(issue, "P3", "platform", "Documentation correction detected.")
if _contains(text, feature):
return _decision(issue, "P3", "product-support", "Feature request detected.")
return _decision(issue, "P2", "product-support", "Routine triage required.")How does the baseline decide?
- The classifier combines the title, body, and customer impact into lowercase text.
- Each broad category maps directly to one priority and routing team.
- The final branch sends unmatched issues to routine product support triage.
- Save app/triage_rules.py.
- Confirm the complete baseline has valid syntax by running this command:
python -m py_compile app/triage_rules.pyWhat should you see?
The command returns to the prompt without an error when the baseline compiles. This proves the classifier is runnable before its decisions face the frozen cases.
Baseline does not compile?
Confirm that each return line is indented beneath its matching condition. Check that every phrase set uses matching braces and quoted strings.
Ask for help with the syntax while preserving the broad baseline: help me repair the syntax in my initial triage classifier
Run the Tier A eval
Tier A checks business rules directly against classify_issue(). Python's unittest subtests keep all ten records in one test while naming each issue in the output.
- Create evals/helpers.py inside the existing evals/ folder.
- Add the fixture loader by pasting this code:
import json
from pathlib import Path
from app.models import Issue
_FIXTURE = Path(__file__).with_name("issues.json")
def load_cases() -> list[tuple[Issue, dict[str, object]]]:
records = json.loads(_FIXTURE.read_text(encoding="utf-8"))
return [(Issue.model_validate(item["issue"]), item["expected"]) for item in records]What does the loader do?
The load_cases() function reads the fixture beside the helper file. It validates every incoming issue through Issue while leaving each expected label dictionary frozen for comparison.
- Create evals/test_tier_a.py inside the same evals/ folder.
- Add the Tier A test by pasting this code:
import unittest
from app.triage_rules import classify_issue
from evals.helpers import load_cases
class TierABaselineEval(unittest.TestCase):
def test_all_labeled_issues(self) -> None:
cases = load_cases()
self.assertEqual(len(cases), 10)
for issue, expected in cases:
with self.subTest(issue_id=issue.issue_id):
decision = classify_issue(issue)
self.assertEqual(decision.priority, expected["priority"])
self.assertEqual(decision.routing_team, expected["routing_team"])
self.assertEqual(decision.requires_human_review, expected["requires_human_review"])
if __name__ == "__main__":
unittest.main()What does Tier A measure?
- The test first confirms that the fixture contains exactly ten cases.
- Each named subtest calls classify_issue() with one validated issue.
- The assertions compare priority, routing team, and human review against the frozen labels.
- Save evals/helpers.py.
- Save evals/test_tier_a.py.
✔️ Awesome, I've got everything!
Great. Double check that all new files are saved before you run Tier A.
ⓧ I'd like to double check the full code
Compare each file with the complete state for this step. The baseline below stays intentionally broad so the evaluation can expose its gaps.
"""Support issue triage service."""from typing import Literal
from pydantic import BaseModel
Priority = Literal["P0", "P1", "P2", "P3"]
RoutingTeam = Literal["security", "payments", "reliability", "platform", "product-support"]
DecisionSource = Literal["baseline", "human_override", "model-assisted", "baseline_fallback"]
class HumanOverride(BaseModel):
priority: Priority
routing_team: RoutingTeam
reason: str
class Issue(BaseModel):
issue_id: str
title: str
body: str
customer_impact: str = ""
human_override: HumanOverride | None = None
class ModelSuggestion(BaseModel):
priority: Priority
routing_team: RoutingTeam
summary: str
class TriageDecision(BaseModel):
issue_id: str
priority: Priority
routing_team: RoutingTeam
requires_human_review: bool
decision_source: DecisionSource
summary: strfrom collections.abc import Iterable
from app.models import Issue, TriageDecision
def _contains(text: str, phrases: Iterable[str]) -> bool:
return any(phrase in text for phrase in phrases)
def _decision(issue: Issue, priority: str, team: str, reason: str) -> TriageDecision:
return TriageDecision(
issue_id=issue.issue_id,
priority=priority,
routing_team=team,
requires_human_review=priority in {"P0", "P1"} or team == "security",
decision_source="baseline",
summary=reason,
)
def classify_issue(issue: Issue) -> TriageDecision:
text = " ".join([issue.title, issue.body, issue.customer_impact]).casefold()
payment = {"payment", "checkout", "invoice", "charge", "billing"}
integration = {"webhook", "api integration", "sdk", "oauth callback"}
performance = {"slow", "latency", "timeout", "performance"}
documentation = {"documentation", "docs", "guide", "typo", "example is wrong"}
feature = {"feature request", "dark mode", "enhancement", "would like"}
if _contains(text, payment):
return _decision(issue, "P2", "payments", "Payment or billing defect detected.")
if _contains(text, integration):
return _decision(issue, "P2", "platform", "Integration defect detected.")
if _contains(text, performance):
return _decision(issue, "P2", "reliability", "Performance degradation detected.")
if _contains(text, documentation):
return _decision(issue, "P3", "platform", "Documentation correction detected.")
if _contains(text, feature):
return _decision(issue, "P3", "product-support", "Feature request detected.")
return _decision(issue, "P2", "product-support", "Routine triage required.")"""Evaluation suite for support issue triage."""import json
from pathlib import Path
from app.models import Issue
_FIXTURE = Path(__file__).with_name("issues.json")
def load_cases() -> list[tuple[Issue, dict[str, object]]]:
records = json.loads(_FIXTURE.read_text(encoding="utf-8"))
return [(Issue.model_validate(item["issue"]), item["expected"]) for item in records][
{"issue":{"issue_id":"ISSUE-001","title":"Production API credentials leaked","body":"Production API credentials leaked in a public log and customer data exposed.","customer_impact":"Credentials and customer data may be accessible externally."},"expected":{"priority":"P0","routing_team":"security","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-002","title":"Checkout and payment processing down for all customers","body":"Every checkout attempt is failing in production.","customer_impact":"All customers are unable to pay."},"expected":{"priority":"P0","routing_team":"payments","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-003","title":"Possible auth bypass","body":"One user may view another profile through an auth bypass.","customer_impact":"Potential unauthorized access requires investigation."},"expected":{"priority":"P1","routing_team":"security","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-004","title":"Multiple customers cannot log in","body":"Login intermittently returns an error.","customer_impact":"Multiple customers are blocked from their accounts."},"expected":{"priority":"P1","routing_team":"reliability","requires_human_review":true}},
{"issue":{"issue_id":"ISSUE-005","title":"Webhook signature rejected after secret rotation","body":"Our webhook integration rejects valid callbacks after rotating the signing secret.","customer_impact":"One integration cannot process callbacks."},"expected":{"priority":"P2","routing_team":"platform","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-006","title":"Dashboard is slow in one region","body":"Dashboard requests take eight seconds for users in one region.","customer_impact":"The dashboard remains usable but is slow."},"expected":{"priority":"P2","routing_team":"reliability","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-007","title":"Invoice total is incorrect","body":"The latest invoice applies a duplicate charge to one account.","customer_impact":"One customer has an incorrect bill."},"expected":{"priority":"P2","routing_team":"payments","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-008","title":"OAuth guide uses the wrong callback parameter","body":"The documentation example is wrong and causes setup confusion.","customer_impact":"Developers can work around the docs error."},"expected":{"priority":"P3","routing_team":"platform","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-009","title":"Feature request for dark mode","body":"Our support team would like dark mode in the admin console.","customer_impact":"No workflow is blocked."},"expected":{"priority":"P3","routing_team":"product-support","requires_human_review":false}},
{"issue":{"issue_id":"ISSUE-010","title":"Routine access report needs audit review","body":"The report succeeded, but an auditor requested a security review.","customer_impact":"No outage is present.","human_override":{"priority":"P1","routing_team":"security","reason":"Auditor requested immediate review."}},"expected":{"priority":"P1","routing_team":"security","requires_human_review":true}}
]import unittest
from app.triage_rules import classify_issue
from evals.helpers import load_cases
class TierABaselineEval(unittest.TestCase):
def test_all_labeled_issues(self) -> None:
cases = load_cases()
self.assertEqual(len(cases), 10)
for issue, expected in cases:
with self.subTest(issue_id=issue.issue_id):
decision = classify_issue(issue)
self.assertEqual(decision.priority, expected["priority"])
self.assertEqual(decision.routing_team, expected["routing_team"])
self.assertEqual(decision.requires_human_review, expected["requires_human_review"])
if __name__ == "__main__":
unittest.main()fastapi==0.147.0
pydantic==2.14.1
uvicorn[standard]==0.54.0
google-genai==2.31.0Before you run the gate, do you think broad category matching can satisfy every frozen label?
- Run all ten records as named Tier A subtests with this command:
python -m unittest -v evals.test_tier_aWhat should you see?
Several straightforward category cases pass. At least one security, routing, or human override case reports a mismatch against its frozen expected label.
That failure is the intended result. The gate has converted a plausible classifier into concrete evidence of where the baseline falls short.
The eval did not reach the assertions?
Confirm that issues.json, helpers.py, and test_tier_a.py are all inside evals/. Confirm that models.py and triage_rules.py are inside app/.
Keep assertion mismatches visible because they are the designed outcome. Ask for help only if the suite cannot load: help me make my Tier A suite load without changing the expected labels
You now have a repeatable gate that catches unsafe assumptions in the first classifier. Next, you will turn those mismatches into ordered guardrails without weakening the frozen labels.
Make the Deterministic Gate Pass
Your labeled eval fixtures have exposed where the naive classifier breaks. The deterministic baseline now needs rules that defend the frozen labels.
The failed cases show that customer impact needs ordered precedence across routing outcomes. A human override must win before automation so explicit judgment is preserved.
In this step, get ready to:
- Create shared decision helpers.
- Place human overrides before automated classification.
- Order impact rules until all 10 frozen labels pass.
Encode shared decisions and overrides
Repeated rule logic can drift when each category builds its own result. Shared helpers keep phrase matching consistent while applying the same review clamp to every automated decision.
- Replace the entire naive implementation in app/triage_rules.py with this helper chunk:
from collections.abc import Iterable
from app.models import Issue, TriageDecision
def _contains(text: str, phrases: Iterable[str]) -> bool:
return any(phrase in text for phrase in phrases)
def _decision(issue: Issue, priority: str, team: str, reason: str) -> TriageDecision:
return TriageDecision(
issue_id=issue.issue_id,
priority=priority,
routing_team=team,
requires_human_review=priority in {"P0", "P1"} or team == "security",
decision_source="baseline",
summary=reason,
)
What do these helpers control?
- The _contains() helper gives every category one phrase-matching rule.
- The _decision() helper applies the review clamp from one place.
- The clamp requires review for P0, P1, or security decisions.
- Save app/triage_rules.py.
- Confirm your code editor reports no Python syntax problem in these helpers.
Seeing a syntax warning?
- Check that both import statements begin at the left edge of app/triage_rules.py.
- Check that each field inside TriageDecision() uses four spaces of indentation.
help me find the syntax problem in my triage helper code.
Explicit judgment needs a path that automation cannot overrule. The first branch returns the supplied priority and routing team before any phrase matching begins.
- Append the classifier setup directly below _decision() by pasting this chunk:
def classify_issue(issue: Issue) -> TriageDecision:
if issue.human_override is not None:
return TriageDecision(
issue_id=issue.issue_id,
priority=issue.human_override.priority,
routing_team=issue.human_override.routing_team,
requires_human_review=True,
decision_source="human_override",
summary=f"Human override: {issue.human_override.reason}",
)
text = " ".join([issue.title, issue.body, issue.customer_impact]).casefold()
critical_security = {"credential leak", "credentials leaked", "private key exposed", "token leaked", "customer data exposed", "actively exploited"}
payment = {"payment", "checkout", "invoice", "charge", "billing"}
global_impact = {"all customers", "sitewide", "global outage", "production down", "service unavailable"}
security = {"auth bypass", "unauthorized access", "suspicious login", "vulnerability", "csrf", "xss"}
widespread = {"multiple customers", "many customers", "multiple tenants", "cannot log in", "unable to log in"}
integration = {"webhook", "api integration", "sdk", "oauth callback"}
performance = {"slow", "latency", "timeout", "performance"}
documentation = {"documentation", "docs", "guide", "typo", "example is wrong"}
feature = {"feature request", "dark mode", "enhancement", "would like"}
How does the classifier start?
- The first branch returns a complete decision whenever human_override contains an explicit judgment.
- The combined text value normalizes the title, body, and customer impact for matching.
- The named phrase sets capture the business signals used by the frozen cases.
- Save app/triage_rules.py.
- Predict which frozen record the override branch now handles correctly.
- Run the current Tier A checkpoint with this command:
python -m unittest -v evals.test_tier_aWhat does this checkpoint test?
The test runner loads the same 10 records from evals/issues.json. Each subtest still compares priority, routing, and review status against its frozen labels.
You should see the automated records still fail because their category checks are incomplete. The override record is no longer among the mismatches.
Seeing an unexpected import or syntax problem?
- Check that classify_issue() begins at the left edge of the file.
- Check that every line inside the override branch uses consistent indentation.
help me debug the incomplete Tier A checkpoint.
Add ordered category checks
Phrase matching becomes policy through its order. High-impact conditions need to return before broader categories can claim the same issue.
- Append the ordered checks below the feature phrase set:
if _contains(text, critical_security):
return _decision(issue, "P0", "security", "Critical security exposure detected.")
if _contains(text, payment) and _contains(text, global_impact):
return _decision(issue, "P0", "payments", "Global payment outage detected.")
if _contains(text, global_impact):
return _decision(issue, "P0", "reliability", "Global service outage detected.")
if _contains(text, security):
return _decision(issue, "P1", "security", "Security investigation required.")
if _contains(text, widespread):
return _decision(issue, "P1", "reliability", "Widespread customer impact detected.")
if _contains(text, payment):
return _decision(issue, "P2", "payments", "Payment or billing defect detected.")
if _contains(text, integration):
return _decision(issue, "P2", "platform", "Integration defect detected.")
if _contains(text, performance):
return _decision(issue, "P2", "reliability", "Performance degradation detected.")
if _contains(text, documentation):
return _decision(issue, "P3", "platform", "Documentation correction detected.")
if _contains(text, feature):
return _decision(issue, "P3", "product-support", "Feature request detected.")
return _decision(issue, "P2", "product-support", "Routine triage required.")Why does the order matter?
- Critical security exposure returns before the broader security rule can assign P1.
- A payment issue with global impact returns before the general outage rule can route it to reliability.
- Specific payment, integration, performance, documentation, and feature rules handle the lower-priority cases.
- The final return gives unmatched issues a guarded routine-triage decision.
- Save app/triage_rules.py.
✔️ I have the ordered rules
The rule engine now contains shared helpers, override precedence, category phrases, ordered checks, and a fallback.
ⓧ I'd like to double check the full code
This reference shows the complete target state for app/triage_rules.py.
from collections.abc import Iterable
from app.models import Issue, TriageDecision
def _contains(text: str, phrases: Iterable[str]) -> bool:
return any(phrase in text for phrase in phrases)
def _decision(issue: Issue, priority: str, team: str, reason: str) -> TriageDecision:
return TriageDecision(
issue_id=issue.issue_id,
priority=priority,
routing_team=team,
requires_human_review=priority in {"P0", "P1"} or team == "security",
decision_source="baseline",
summary=reason,
)
def classify_issue(issue: Issue) -> TriageDecision:
if issue.human_override is not None:
return TriageDecision(
issue_id=issue.issue_id,
priority=issue.human_override.priority,
routing_team=issue.human_override.routing_team,
requires_human_review=True,
decision_source="human_override",
summary=f"Human override: {issue.human_override.reason}",
)
text = " ".join([issue.title, issue.body, issue.customer_impact]).casefold()
critical_security = {"credential leak", "credentials leaked", "private key exposed", "token leaked", "customer data exposed", "actively exploited"}
payment = {"payment", "checkout", "invoice", "charge", "billing"}
global_impact = {"all customers", "sitewide", "global outage", "production down", "service unavailable"}
security = {"auth bypass", "unauthorized access", "suspicious login", "vulnerability", "csrf", "xss"}
widespread = {"multiple customers", "many customers", "multiple tenants", "cannot log in", "unable to log in"}
integration = {"webhook", "api integration", "sdk", "oauth callback"}
performance = {"slow", "latency", "timeout", "performance"}
documentation = {"documentation", "docs", "guide", "typo", "example is wrong"}
feature = {"feature request", "dark mode", "enhancement", "would like"}
if _contains(text, critical_security):
return _decision(issue, "P0", "security", "Critical security exposure detected.")
if _contains(text, payment) and _contains(text, global_impact):
return _decision(issue, "P0", "payments", "Global payment outage detected.")
if _contains(text, global_impact):
return _decision(issue, "P0", "reliability", "Global service outage detected.")
if _contains(text, security):
return _decision(issue, "P1", "security", "Security investigation required.")
if _contains(text, widespread):
return _decision(issue, "P1", "reliability", "Widespread customer impact detected.")
if _contains(text, payment):
return _decision(issue, "P2", "payments", "Payment or billing defect detected.")
if _contains(text, integration):
return _decision(issue, "P2", "platform", "Integration defect detected.")
if _contains(text, performance):
return _decision(issue, "P2", "reliability", "Performance degradation detected.")
if _contains(text, documentation):
return _decision(issue, "P3", "platform", "Documentation correction detected.")
if _contains(text, feature):
return _decision(issue, "P3", "product-support", "Feature request detected.")
return _decision(issue, "P2", "product-support", "Routine triage required.")
What should match?
This reference preserves one precedence chain from the override branch through the routine fallback. Compare indentation carefully because every category check remains inside classify_issue().
Run Tier A against frozen labels
The 10 fixture labels remain unchanged. A passing result now proves that the implementation meets the agreed policy without moving the target.
- Predict whether the same frozen records now expose a gap or complete cleanly.
- Run the full Tier A gate again with this command:
python -m unittest -v evals.test_tier_aWhat does the final run prove?
The suite sends all 10 labeled issues through classify_issue(). It checks priority, routing team, and review status for every record.
You should see the test method pass. The run ends with OK.
Strong result. All 10 frozen cases now pass through one deterministic policy.
Tier A still showing a mismatch?
- Check that the critical security rule appears before the general security rule.
- Check that the combined payment outage rule appears before the general outage rule.
- Confirm that evals/issues.json still contains the original frozen labels.
help me diagnose the remaining Tier A mismatch without changing the fixture.
Your deterministic guardrail is locked in. Next up, controlled model responses will test downgrades, provider failures, and overrides.
Test Model-Assisted Resilience
Your deterministic baseline now labels all 10 support issues correctly. That gate protects the business rules before model assistance enters the decision path.
Tier B adds a controlled model stub that can force an unsafe downgrade or a provider failure. You will use safety clamps to preserve priority rules while fallback behavior keeps requests usable during provider failures.
In this step, get ready to:
- Define the provider boundary for structured model suggestions.
- Protect deterministic decisions from unsafe model behavior.
- Run 10 requests through a stubbed provider without a network call.
Define the model-provider boundary
A provider boundary gives the orchestrator one stable interface for model suggestions. The concrete Gemini implementation returns validated JSON that matches the existing ModelSuggestion schema.
Why use a provider boundary?
The boundary keeps vendor-specific SDK code inside one file. Tests can supply a stub that behaves like the provider without credentials or network access.
This separation makes rare failures repeatable. It also gives the orchestrator one place to enforce business guardrails.
- Create app/model_provider.py inside the existing app/ folder using your editor.
- Add the provider imports plus the protocol by pasting this code:
from typing import Protocol
from google import genai
from google.genai import types
from app.models import Issue, ModelSuggestion, TriageDecision
class ModelProvider(Protocol):
def suggest(self, issue: Issue, baseline: TriageDecision) -> ModelSuggestion:
"""Return a structured triage suggestion."""What does this code do?
- The ModelProvider protocol defines the contract used by the orchestrator.
- The suggest() method accepts the original issue plus its deterministic baseline.
- The return type requires every provider to produce a validated ModelSuggestion.
- Save app/model_provider.py.
- Compare the protocol signature with the code block.
You should see one suggest() method with typed inputs plus a ModelSuggestion return type.
Protocol definition showing a problem?
Confirm app/models.py still defines Issue. Check that the same file defines ModelSuggestion plus TriageDecision.
Check that Protocol is imported from typing.
help me debug the provider protocol in app/model_provider.py.
- Add the concrete provider directly below the ModelProvider protocol by pasting this code:
class GeminiModelProvider:
def __init__(self, model: str = "gemini-3.8-flash") -> None:
self._model = model
self._client = genai.Client()
def suggest(self, issue: Issue, baseline: TriageDecision) -> ModelSuggestion:
prompt = (
"Review this support issue and deterministic baseline. Suggest a priority, routing team, and concise summary. "
"Escalate when supported, but do not invent facts.\n\n"
f"Issue:\n{issue.model_dump_json()}\n\nBaseline:\n{baseline.model_dump_json()}"
)
response = self._client.models.generate_content(
model=self._model,
contents=prompt,
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_json_schema=ModelSuggestion.model_json_schema(),
),
)
if not response.text:
raise ValueError("Gemini returned no text response")
return ModelSuggestion.model_validate_json(response.text)How does the provider stay structured?
- The constructor stores the stable gemini-3.8-flash model identifier.
- The client reads the existing Gemini credential from the activated environment.
- The prompt includes the issue plus the deterministic baseline as serialized data.
- The response configuration requests JSON matching the ModelSuggestion schema.
- The final validation converts provider text into a typed suggestion.
- Save app/model_provider.py.
- Compare the end of the file with the new provider class.
The file should end with ModelSuggestion.model_validate_json(response.text). This confirms model output crosses the boundary through schema validation.
Provider code showing an import problem?
Confirm the activated environment still contains the pinned Google Gen AI SDK dependency. Check each Google import against the supplied code.
Keep both Google imports inside app/model_provider.py because this file owns the SDK boundary.
help me debug the Gemini provider imports without moving SDK code outside app/model_provider.py.
✔️ Awesome, I've got everything!
Your provider protocol plus the concrete Gemini implementation are saved in app/model_provider.py.
ⓧ I'd like to double check the full code
Compare your complete app/model_provider.py file with this reference.
from typing import Protocol
from google import genai
from google.genai import types
from app.models import Issue, ModelSuggestion, TriageDecision
class ModelProvider(Protocol):
def suggest(self, issue: Issue, baseline: TriageDecision) -> ModelSuggestion:
"""Return a structured triage suggestion."""
class GeminiModelProvider:
def __init__(self, model: str = "gemini-3.8-flash") -> None:
self._model = model
self._client = genai.Client()
def suggest(self, issue: Issue, baseline: TriageDecision) -> ModelSuggestion:
prompt = (
"Review this support issue and deterministic baseline. Suggest a priority, routing team, and concise summary. "
"Escalate when supported, but do not invent facts.\n\n"
f"Issue:\n{issue.model_dump_json()}\n\nBaseline:\n{baseline.model_dump_json()}"
)
response = self._client.models.generate_content(
model=self._model,
contents=prompt,
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_json_schema=ModelSuggestion.model_json_schema(),
),
)
if not response.text:
raise ValueError("Gemini returned no text response")
return ModelSuggestion.model_validate_json(response.text)What should match?
The complete file contains one protocol plus one concrete provider. All Google SDK imports remain inside this boundary.
- Confirm the deterministic gate still passes by running:
python -m unittest -v evals.test_tier_aWhat should you see?
The Tier A suite should complete successfully. This confirms the provider boundary did not alter the frozen deterministic decisions.
Did Tier A stop passing?
The new provider file does not require changes to app/triage_rules.py. Restore any accidental edits to the deterministic rules.
help me find why Tier A changed after adding app/model_provider.py.
Guard the orchestration path
The orchestrator combines the trusted baseline with an optional provider suggestion. Its priority ranking permits escalation while blocking any attempted downgrade.
- Create app/orchestrator.py inside the existing app/ folder using your editor.
- Add the imports plus the priority ranking by pasting this code:
from app.model_provider import GeminiModelProvider, ModelProvider
from app.models import Issue, TriageDecision
from app.triage_rules import classify_issue
_PRIORITY_RANK = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}Why rank priorities?
Lower numbers represent more urgent priorities. Comparing these numbers gives the orchestrator a deterministic escalation rule.
A suggestion can move from P2 to P1. A suggestion cannot move from P0 to P3.
- Save app/orchestrator.py.
- Compare the four priority values with the code block.
You should see P0 ranked at 0. You should see P3 ranked at 3.
Priority comparison looking backwards?
Confirm higher urgency uses a lower number. The escalation check depends on that ordering.
help me verify the priority ranking in app/orchestrator.py.
- Add the orchestrator directly below _PRIORITY_RANK by pasting this code:
class TriageOrchestrator:
def __init__(self, provider: ModelProvider) -> None:
self._provider = provider
def triage(self, issue: Issue) -> TriageDecision:
baseline = classify_issue(issue)
if baseline.decision_source == "human_override":
return baseline
try:
suggestion = self._provider.suggest(issue, baseline)
except Exception:
return baseline.model_copy(update={
"decision_source": "baseline_fallback",
"summary": f"{baseline.summary} Model provider unavailable.",
})
is_escalation = _PRIORITY_RANK[suggestion.priority] < _PRIORITY_RANK[baseline.priority]
priority = suggestion.priority if is_escalation else baseline.priority
team = suggestion.routing_team if is_escalation else baseline.routing_team
return baseline.model_copy(update={
"priority": priority,
"routing_team": team,
"requires_human_review": baseline.requires_human_review or priority in {"P0", "P1"} or team == "security",
"decision_source": "model-assisted",
"summary": suggestion.summary,
})
def build_orchestrator() -> TriageOrchestrator:
return TriageOrchestrator(GeminiModelProvider())How are unsafe outcomes blocked?
- The baseline runs before any provider call.
- A human override returns immediately without consulting the provider.
- A provider exception returns the baseline with a baseline_fallback source.
- The ranking comparison accepts only a more urgent suggested priority.
- The final review rule preserves review for urgent or security decisions.
- Save app/orchestrator.py.
- Compare the triage() branches with the code block.
The method should contain an early override return plus an exception fallback. It should finish with a guarded model-assisted decision.
Orchestrator file showing a type problem?
Confirm TriageOrchestrator receives a ModelProvider. Check that classify_issue is imported from app.triage_rules.
Check every model_copy update key against the fields in TriageDecision.
help me debug the guarded branches in TriageOrchestrator.triage().
✔️ Awesome, I've got everything!
Your orchestrator now combines the baseline with guarded provider suggestions.
ⓧ I'd like to double check the full code
Compare your complete app/orchestrator.py file with this reference.
from app.model_provider import GeminiModelProvider, ModelProvider
from app.models import Issue, TriageDecision
from app.triage_rules import classify_issue
_PRIORITY_RANK = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}
class TriageOrchestrator:
def __init__(self, provider: ModelProvider) -> None:
self._provider = provider
def triage(self, issue: Issue) -> TriageDecision:
baseline = classify_issue(issue)
if baseline.decision_source == "human_override":
return baseline
try:
suggestion = self._provider.suggest(issue, baseline)
except Exception:
return baseline.model_copy(update={
"decision_source": "baseline_fallback",
"summary": f"{baseline.summary} Model provider unavailable.",
})
is_escalation = _PRIORITY_RANK[suggestion.priority] < _PRIORITY_RANK[baseline.priority]
priority = suggestion.priority if is_escalation else baseline.priority
team = suggestion.routing_team if is_escalation else baseline.routing_team
return baseline.model_copy(update={
"priority": priority,
"routing_team": team,
"requires_human_review": baseline.requires_human_review or priority in {"P0", "P1"} or team == "security",
"decision_source": "model-assisted",
"summary": suggestion.summary,
})
def build_orchestrator() -> TriageOrchestrator:
return TriageOrchestrator(GeminiModelProvider())What should match?
The full file contains the priority ranking plus one orchestrator class. The final factory builds the production provider.
- Protect the existing baseline from accidental changes by running:
python -m unittest -v evals.test_tier_aWhat does this check prove?
Tier A should still complete successfully against the unchanged fixture. The new orchestration layer sits above the deterministic rules.
Did the baseline regress?
Check git diff in your editor for accidental changes to app/triage_rules.py or evals/issues.json.
help me trace a Tier A regression without changing the frozen expected labels.
Create the no-network Tier B suite
Tier B replaces the production provider with a stub. The same 10 issues now exercise override bypasses plus downgrade protection plus provider fallback through TriageOrchestrator.
Why a stub instead of live Gemini?
A stub produces the same behavior on every run. It can also force rare failures without consuming API quota.
This makes the merge gate repeatable. It also keeps credentials out of automated evaluation runs.
- Create evals/test_tier_b.py inside the existing evals/ folder using your editor.
- Add the imports plus the stub provider by pasting this code:
import unittest
from app.models import Issue, ModelSuggestion, TriageDecision
from app.orchestrator import TriageOrchestrator
from evals.helpers import load_cases
class StubModelProvider:
def __init__(self) -> None:
self.calls: list[str] = []
def suggest(self, issue: Issue, baseline: TriageDecision) -> ModelSuggestion:
self.calls.append(issue.issue_id)
if issue.issue_id == "ISSUE-001":
return ModelSuggestion(priority="P3", routing_team="product-support", summary="Unsafe downgrade attempt.")
if issue.issue_id == "ISSUE-005":
raise RuntimeError("Simulated provider failure")
return ModelSuggestion(priority=baseline.priority, routing_team=baseline.routing_team, summary=f"Stub-reviewed {issue.issue_id}.")What failures does the stub force?
- The calls list records every issue sent to the provider.
- The ISSUE-001 branch attempts to downgrade a critical security issue to P3.
- The ISSUE-005 branch simulates a provider failure.
- Every other provider call returns the deterministic priority plus routing team.
- Save evals/test_tier_b.py.
- Compare the two forced branches with the code block.
You should see one unsafe downgrade attempt for ISSUE-001. You should also see one simulated failure for ISSUE-005.
Stub types not matching?
Confirm suggest() accepts an Issue plus a TriageDecision. Its successful branches must return a ModelSuggestion.
help me fix the StubModelProvider signature or return values.
- Add the Tier B test directly below StubModelProvider by pasting this code:
class TierBOrchestratorEval(unittest.TestCase):
def test_all_requests_preserve_expected_outcomes(self) -> None:
provider = StubModelProvider()
orchestrator = TriageOrchestrator(provider)
cases = load_cases()
self.assertEqual(len(cases), 10)
for issue, expected in cases:
with self.subTest(issue_id=issue.issue_id):
decision = orchestrator.triage(issue)
self.assertEqual(decision.priority, expected["priority"])
self.assertEqual(decision.routing_team, expected["routing_team"])
self.assertEqual(decision.requires_human_review, expected["requires_human_review"])
expected_source = "baseline_fallback" if issue.issue_id == "ISSUE-005" else "human_override" if issue.issue_id == "ISSUE-010" else "model-assisted"
self.assertEqual(decision.decision_source, expected_source)
self.assertEqual(len(provider.calls), 9)
self.assertNotIn("ISSUE-010", provider.calls)
if __name__ == "__main__":
unittest.main()What does Tier B prove?
- The suite loads exactly 10 frozen issues.
- Every request must preserve the expected priority plus routing team plus review flag.
- The provider failure must produce baseline_fallback.
- The human override must preserve human_override as its decision source.
- The nine-call assertion proves ISSUE-010 bypasses the provider.
- Save evals/test_tier_b.py.
✔️ Awesome, I've got everything!
Your Tier B suite now covers all 10 issues through the guarded orchestration path.
ⓧ I'd like to double check the full code
Compare your complete evals/test_tier_b.py file with this reference.
import unittest
from app.models import Issue, ModelSuggestion, TriageDecision
from app.orchestrator import TriageOrchestrator
from evals.helpers import load_cases
class StubModelProvider:
def __init__(self) -> None:
self.calls: list[str] = []
def suggest(self, issue: Issue, baseline: TriageDecision) -> ModelSuggestion:
self.calls.append(issue.issue_id)
if issue.issue_id == "ISSUE-001":
return ModelSuggestion(priority="P3", routing_team="product-support", summary="Unsafe downgrade attempt.")
if issue.issue_id == "ISSUE-005":
raise RuntimeError("Simulated provider failure")
return ModelSuggestion(priority=baseline.priority, routing_team=baseline.routing_team, summary=f"Stub-reviewed {issue.issue_id}.")
class TierBOrchestratorEval(unittest.TestCase):
def test_all_requests_preserve_expected_outcomes(self) -> None:
provider = StubModelProvider()
orchestrator = TriageOrchestrator(provider)
cases = load_cases()
self.assertEqual(len(cases), 10)
for issue, expected in cases:
with self.subTest(issue_id=issue.issue_id):
decision = orchestrator.triage(issue)
self.assertEqual(decision.priority, expected["priority"])
self.assertEqual(decision.routing_team, expected["routing_team"])
self.assertEqual(decision.requires_human_review, expected["requires_human_review"])
expected_source = "baseline_fallback" if issue.issue_id == "ISSUE-005" else "human_override" if issue.issue_id == "ISSUE-010" else "model-assisted"
self.assertEqual(decision.decision_source, expected_source)
self.assertEqual(len(provider.calls), 9)
self.assertNotIn("ISSUE-010", provider.calls)
if __name__ == "__main__":
unittest.main()What should match?
The complete file contains one stub plus one test case. The final assertions require nine provider calls with no call for ISSUE-010.
Before you run Tier B, which guard do you expect to stop the ISSUE-001 downgrade?
- Run all 10 requests through the stubbed orchestration path with this command:
python -m unittest -v evals.test_tier_bWhat should you see?
The Tier B suite should complete successfully. Its assertions confirm the unsafe downgrade was blocked plus the provider failure used the baseline fallback.
The suite also confirms the human override bypassed the provider. No Gemini network call occurred because the orchestrator received StubModelProvider.
Strong result. Your model-assisted path now survives the three failures a client is most likely to challenge during a technical review.
Tier B not passing?
If ISSUE-001 fails, compare the priority ranking plus the is_escalation condition. A suggested P3 must not replace the P0 baseline.
If ISSUE-005 fails, confirm the provider call sits inside the try block. The exception branch must return baseline_fallback.
If the provider call count is 10, confirm the human-override return appears before self._provider.suggest().
help me diagnose which Tier B guard is failing.
Your provider boundary plus resilience gate are ready. Next, you will expose the guarded flow through FastAPI before enforcing the same checks with GitHub Actions.
Wire the API, Architecture, and CI Gate
Your deterministic rules and stubbed resilience tests now protect the triage decisions. A client handoff still needs a real HTTP path that proves the complete service works.
In this step, FastAPI exposes the guarded flow. A Mermaid C4 diagram records the boundaries. GitHub Actions turns the architecture policy plus both eval tiers into a merge gate.
In this step, get ready to:
- Expose the guarded triage flow through a live API endpoint.
- Document the service boundary in the C4 container view.
- Run architecture checks plus both eval tiers in one CI job.
Expose and call the triage endpoint
The existing orchestrator already owns the safe decision flow. The API layer validates each request as an Issue before returning a TriageDecision.
- Create app/main.py in the code editor from earlier.
- Add the API endpoint by pasting this code:
from fastapi import FastAPI
from app.models import Issue, TriageDecision
from app.orchestrator import build_orchestrator
app = FastAPI(title="Support Issue Triage Service")
orchestrator = build_orchestrator()
@app.post("/triage", response_model=TriageDecision)
def triage_issue(issue: Issue) -> TriageDecision:
return orchestrator.triage(issue)What does this API layer do?
- The global orchestrator reuses the provider boundary plus the deterministic guardrails from earlier.
- The POST /triage route validates the request against Issue.
- The response_model requires the response to match TriageDecision.
- Save app/main.py.
- Start the API with the existing .env file by running:
python -m uvicorn app.main:app --reload --env-file .envWhat does this command start?
Uvicorn loads the FastAPI application from app/main.py. The environment file supplies the existing Gemini credential without printing it.
The server keeps this terminal occupied while it waits for requests. Auto-reload applies saved Python changes during development.
API server not starting?
Confirm you ran the command from the repository containing app/main.py.
Confirm the existing .env file still contains the required variable name without displaying its value.
Ask for help with the startup problem: help me diagnose why my FastAPI triage service does not start
✔️ Awesome, I've got everything!
Your FastAPI application is saved. The guarded orchestrator is now reachable through POST /triage.
ⓧ I'd like to double check the full code
- Compare app/main.py with this complete file:
from fastapi import FastAPI
from app.models import Issue, TriageDecision
from app.orchestrator import build_orchestrator
app = FastAPI(title="Support Issue Triage Service")
orchestrator = build_orchestrator()
@app.post("/triage", response_model=TriageDecision)
def triage_issue(issue: Issue) -> TriageDecision:
return orchestrator.triage(issue)What should match?
The endpoint must call the global orchestrator. Direct provider access here would break the architecture boundary.
The service is listening. A small caller now provides a repeatable client demonstration without exposing the API key in a command.
- Create scripts/call_triage.py in the code editor from earlier.
- Add the demonstration request by pasting this code:
import json
from urllib.request import Request, urlopen
payload = {
"issue_id": "DEMO-001",
"title": "Checkout and payment processing down for all customers",
"body": "Every checkout attempt is failing in production.",
"customer_impact": "All customers are unable to pay.",
}
request = Request(
"http://127.0.0.1:8000/triage",
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
with urlopen(request, timeout=30) as response:
result = json.loads(response.read().decode("utf-8"))
print(json.dumps(result, indent=2))How does the caller work?
- The payload represents a global checkout outage with the demonstration ID DEMO-001.
- The Request sends JSON to the local POST /triage endpoint.
- The final line formats the returned decision so every guarded field is easy to inspect.
- Save scripts/call_triage.py.
- Open a second terminal in the same repository.
- Activate the existing .venv using the same Linux activation command from Step 1.
This next check sends one supplied demonstration issue through the original Gemini key. A paid key can incur token charges for this single request.
Before you send the request, consider this question: can the model lower the guarded P0 decision.
- Send the live request by running:
python scripts/call_triage.pyWhat should I see?
You should see a formatted JSON response for DEMO-001. It includes priority, routing_team, requires_human_review, decision_source, plus summary.
The guarded result keeps priority P0. It routes the outage to payments with human review required.
Live request not returning JSON?
Confirm the first terminal still has the Uvicorn server running. The caller cannot reach the endpoint after that process stops.
Confirm the second terminal uses the existing virtual environment. Keep the API key out of screenshots plus help requests.
Ask for help without sharing credentials: help me debug why my local triage caller cannot get a JSON response
- Return to the terminal running Uvicorn.
- Stop the server by pressing Ctrl+C.
✔️ Awesome, I've got everything!
That live response proves the HTTP path reaches the guarded orchestrator. Your demonstration caller can now reproduce the client-facing flow.
ⓧ I'd like to double check the full code
- Compare scripts/call_triage.py with this complete file:
import json
from urllib.request import Request, urlopen
payload = {
"issue_id": "DEMO-001",
"title": "Checkout and payment processing down for all customers",
"body": "Every checkout attempt is failing in production.",
"customer_impact": "All customers are unable to pay.",
}
request = Request(
"http://127.0.0.1:8000/triage",
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
with urlopen(request, timeout=30) as response:
result = json.loads(response.read().decode("utf-8"))
print(json.dumps(result, indent=2))What should match?
The caller must target http://127.0.0.1:8000/triage with the supplied DEMO-001 payload. It prints the decoded response without reading or displaying the Gemini key.
Document the service boundary
The live request proves the runtime path. The architecture view makes that path defensible by showing where deterministic rules, model assistance, evaluations, plus CI each belong.
- Create docs/architecture.md in the code editor from earlier.
- Add the complete container view by pasting this content:
# Support Issue Triage Architecture
```mermaid
C4Container
title Support Issue Triage Service with Evaluation Gate
Person(engineer, "AI Solutions Engineer", "Changes rules and service code")
System_Boundary(service, "Support Issue Triage Service") {
Container(api, "FastAPI API", "Python, FastAPI", "Accepts POST /triage")
Container(orchestrator, "Triage Orchestrator", "Python", "Combines guarded decisions")
Container(rules, "Deterministic Rule Engine", "Python", "Owns triage guardrails")
Container(provider, "Gemini Model Provider", "Google Gen AI SDK", "Requests structured suggestions")
Container(evals, "Two-Tier Eval Harness", "unittest, JSON", "Runs regression cases")
}
System_Ext(gemini, "Gemini 3.8 Flash", "Returns structured suggestions")
System_Ext(actions, "GitHub Actions", "Runs architecture and eval checks")
Rel(engineer, api, "Invokes")
Rel(api, orchestrator, "Sends validated issue")
Rel(orchestrator, rules, "Gets baseline")
Rel(orchestrator, provider, "Requests suggestion")
Rel(provider, gemini, "Calls generate_content")
Rel(evals, rules, "Runs Tier A")
Rel(evals, orchestrator, "Runs Tier B with stub")
Rel(actions, evals, "Runs regression gate")
Rel(actions, orchestrator, "Enforces import boundary")
```What does the diagram prove?
The FastAPI API reaches Gemini only through the Triage Orchestrator plus Gemini Model Provider. The deterministic rule engine remains the source of guarded baseline decisions.
The eval harness tests rules directly in Tier A. It tests orchestration through a stub in Tier B while GitHub Actions enforces both regression behavior plus the import boundary.
- Save docs/architecture.md.
- Return to the Mermaid preview used in Part 1.
- Confirm the rendered service boundary contains five internal containers.
- Confirm the rendered diagram shows the external Gemini plus GitHub Actions systems.
You should see separate relationships for Tier A, Tier B, the live provider call, the regression gate, plus the import policy. This view shows that model access has one controlled route.
Diagram not rendering?
Confirm the opening Mermaid fence begins directly above C4Container. Confirm the closing fence appears after the final relationship.
Ask for help with the diagram syntax: help me find the syntax problem in my Mermaid C4 container diagram
✔️ Awesome, I've got everything!
Your architecture view now shows the runtime boundary plus the two-tier regression gate in one client-ready diagram.
ⓧ I'd like to double check the full code
- Compare docs/architecture.md with this complete file:
# Support Issue Triage Architecture
```mermaid
C4Container
title Support Issue Triage Service with Evaluation Gate
Person(engineer, "AI Solutions Engineer", "Changes rules and service code")
System_Boundary(service, "Support Issue Triage Service") {
Container(api, "FastAPI API", "Python, FastAPI", "Accepts POST /triage")
Container(orchestrator, "Triage Orchestrator", "Python", "Combines guarded decisions")
Container(rules, "Deterministic Rule Engine", "Python", "Owns triage guardrails")
Container(provider, "Gemini Model Provider", "Google Gen AI SDK", "Requests structured suggestions")
Container(evals, "Two-Tier Eval Harness", "unittest, JSON", "Runs regression cases")
}
System_Ext(gemini, "Gemini 3.8 Flash", "Returns structured suggestions")
System_Ext(actions, "GitHub Actions", "Runs architecture and eval checks")
Rel(engineer, api, "Invokes")
Rel(api, orchestrator, "Sends validated issue")
Rel(orchestrator, rules, "Gets baseline")
Rel(orchestrator, provider, "Requests suggestion")
Rel(provider, gemini, "Calls generate_content")
Rel(evals, rules, "Runs Tier A")
Rel(evals, orchestrator, "Runs Tier B with stub")
Rel(actions, evals, "Runs regression gate")
Rel(actions, orchestrator, "Enforces import boundary")
```What should match?
The full diagram must include every runtime container plus both external systems. Its relationships must show the eval harness reaching rules plus the orchestrator through separate tiers.
Enforce the boundary in CI
The diagram communicates the intended architecture. An executable import policy prevents future changes from bypassing it.
The policy reads Python syntax trees instead of searching raw text. This lets it inspect actual imports across every Python file under app/.
- Create architecture/check_imports.py in the code editor from earlier.
- Add the imports plus policy constants by pasting this first section:
import ast
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
APP = ROOT / "app"
PROVIDER = "app/model_provider.py"
ORCHESTRATOR = "app/orchestrator.py"
def modules(tree: ast.AST) -> list[str]:
found: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.Import):
found.extend(alias.name for alias in node.names)
elif isinstance(node, ast.ImportFrom) and node.module:
found.append(node.module)
return foundHow does the import scanner work?
- The path constants define the application folder plus the two files allowed to cross protected boundaries.
- The modules() helper walks the parsed syntax tree to collect both import styles.
- Parsing imports structurally avoids matching module names inside comments or ordinary strings.
- Save architecture/check_imports.py.
- Confirm architecture/check_imports.py appears in the editor file sidebar.
File missing from the sidebar?
Confirm the file sits inside the repository's architecture/ folder. The final path must be architecture/check_imports.py.
Ask for help locating the policy file: help me correct the location of my architecture policy script
- Append the policy checks directly below return found by pasting this section:
def main() -> None:
violations: list[str] = []
for path in sorted(APP.rglob("*.py")):
relative = path.relative_to(ROOT).as_posix()
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for module in modules(tree):
if module.split(".")[0] == "google" and relative != PROVIDER:
violations.append(f"{relative}: only {PROVIDER} may import Google SDK modules")
if module == "app.model_provider" and relative != ORCHESTRATOR:
violations.append(f"{relative}: only {ORCHESTRATOR} may import app.model_provider")
if violations:
raise SystemExit("\n".join(violations))
print("Architecture import policy passed.")
if __name__ == "__main__":
main()What does the policy enforce?
- Only app/model_provider.py may import Google SDK modules.
- Only app/orchestrator.py may import app.model_provider.
- Any forbidden edge stops the script with the file path plus the violated rule.
- Save architecture/check_imports.py.
- Check the current application imports by running:
python architecture/check_imports.pyWhat should I see?
You should see Architecture import policy passed. in the terminal. This confirms the current API reaches the model provider only through the orchestrator.
Policy reporting a violation?
Read the reported file path first. A Google SDK import belongs only in app/model_provider.py.
An app.model_provider import belongs only in app/orchestrator.py.
Ask for help interpreting the dependency edge: help me diagnose the forbidden import reported by my architecture policy
✔️ Awesome, I've got everything!
Your executable architecture policy accepts the current dependency graph. Future boundary shortcuts now produce a concrete failure.
ⓧ I'd like to double check the full code
- Compare architecture/check_imports.py with this complete file:
import ast
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
APP = ROOT / "app"
PROVIDER = "app/model_provider.py"
ORCHESTRATOR = "app/orchestrator.py"
def modules(tree: ast.AST) -> list[str]:
found: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.Import):
found.extend(alias.name for alias in node.names)
elif isinstance(node, ast.ImportFrom) and node.module:
found.append(node.module)
return found
def main() -> None:
violations: list[str] = []
for path in sorted(APP.rglob("*.py")):
relative = path.relative_to(ROOT).as_posix()
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for module in modules(tree):
if module.split(".")[0] == "google" and relative != PROVIDER:
violations.append(f"{relative}: only {PROVIDER} may import Google SDK modules")
if module == "app.model_provider" and relative != ORCHESTRATOR:
violations.append(f"{relative}: only {ORCHESTRATOR} may import app.model_provider")
if violations:
raise SystemExit("\n".join(violations))
print("Architecture import policy passed.")
if __name__ == "__main__":
main()What should match?
The policy must scan every Python file under app/. Its two exceptions must name the provider plus orchestrator paths exactly.
The policy passes locally. The workflow now places that structural check before both behavior tiers on every pushed change plus pull request.
- Create .github/workflows/architecture-check.yml in the code editor from earlier.
- Add the combined CI job by pasting this workflow:
name: Architecture and Eval Checks
on:
push:
pull_request:
permissions:
contents: read
jobs:
check:
runs-on: ubuntu-latest
steps:
- name: Check out repository
uses: actions/checkout@v7
- name: Set up Python
uses: actions/setup-python@v7
with:
python-version: "3.11"
- name: Install dependencies
run: python -m pip install -r requirements.txt
- name: Enforce architecture boundary
run: python architecture/check_imports.py
- name: Run Tier A and Tier B evals
run: python -m unittest -v evals.test_tier_a evals.test_tier_bHow does the gate run?
- The workflow runs for repository pushes plus pull requests.
- The job checks out the repository before setting up Python 3.11.
- The architecture policy runs before the Tier A plus Tier B eval modules.
- A failing policy or eval command prevents the job from completing successfully.
- Save .github/workflows/architecture-check.yml.
✔️ Awesome, I've got everything!
Your workflow now treats architecture plus behavior as one release gate. Both checks must pass in the same job.
ⓧ I'd like to double check the full code
- Compare .github/workflows/architecture-check.yml with this complete file:
name: Architecture and Eval Checks
on:
push:
pull_request:
permissions:
contents: read
jobs:
check:
runs-on: ubuntu-latest
steps:
- name: Check out repository
uses: actions/checkout@v7
- name: Set up Python
uses: actions/setup-python@v7
with:
python-version: "3.11"
- name: Install dependencies
run: python -m pip install -r requirements.txt
- name: Enforce architecture boundary
run: python architecture/check_imports.py
- name: Run Tier A and Tier B evals
run: python -m unittest -v evals.test_tier_a evals.test_tier_bWhat should match?
The workflow must run the import policy before the combined Tier A plus Tier B command. It must use the pinned Python version plus dependencies from requirements.txt.
Before you run the combined gate, consider which failure would prevent the second command from running.
- Run the complete local gate with this command:
python architecture/check_imports.py && python -m unittest -v evals.test_tier_a evals.test_tier_bWhat should I see?
You should first see Architecture import policy passed..
Tier A plus Tier B should then complete all 20 fixture evaluations. The unittest run ends with OK.
Combined gate not green?
If only the first command runs, fix its reported import edge before retrying. The second command starts only after the architecture policy succeeds.
If an eval fails, use its issue ID to decide whether the mismatch comes from deterministic classification or orchestration behavior.
Ask for help with the failing layer: help me diagnose which part of my combined architecture and eval gate failed
- Commit the completed files with Git using the same commit process from Part 1.
- Push the current Part 2 branch using the same push process from Part 1.
Before you inspect the remote run, consider whether the architecture policy plus both eval tiers will execute inside one job.
- Return to this repository on GitHub.
- Open the workflow run for the branch you just pushed.
- Confirm the Architecture and Eval Checks workflow completes successfully.
You should see one successful job that installs the dependencies before running the import policy plus both eval tiers. This proves the same gate works locally and on every repository change.
That is the client handoff complete. Your repository now demonstrates a live guarded decision, a documented architecture boundary, plus a merge-blocking regression gate.
Secret mission
Trigger One Architecture Boundary Failure
A structural policy earns trust after it catches a realistic shortcut. You will temporarily place a forbidden SDK import in the API layer, inspect the rejected dependency edge, restore the approved boundary, and prove that all 20 eval requests remain green.
Clean Up Your Resources
Clean Up Your Resources
Your local evals use the supplied model stub. Your GitHub Actions gate uses the same stub without creating ongoing model cost.
Decide whether to keep the Part 2 branch ready for client work, pause the local processes, or remove the evaluation extension.
Cost warning
Free-tier input and output tokens for gemini-3.8-flash are free of charge. A paid project bills model use by token volume.
The local server has no idle cloud charge. Charges only arise when a paid key handles another request.
Resources you used:
- The local Part 2 Git branch containing your evaluation extension.
- The remote Part 2 branch on GitHub if you pushed it.
- The versioned Part 2 assets in app/, evals/, architecture/, docs/, scripts/, .github/workflows/architecture-check.yml, and requirements.txt.
- The .venv virtual environment inside the repository.
- A local Uvicorn process serving the FastAPI API if it is still running.
Keep everything running
No action is needed. Choose this if you plan to reuse the triage gate for client rules or future regression cases.
- Keep your local Part 2 branch so the source code remains available.
- Keep the remote Part 2 branch so GitHub can rerun the combined architecture and evaluation gate.
- Keep .venv so the pinned dependencies remain ready for another local run.
- Leave Uvicorn running only while you are actively testing the endpoint.
- Remember that another run of scripts/call_triage.py sends another live model request.
Pause - I'll come back to this later
Pause the local processes to free memory while preserving every branch, fixture, test, and workflow.
- Switch back to the terminal running Uvicorn from earlier if the server is still active.
- Press Ctrl+C to stop the local server.
You should see the shell prompt return when Uvicorn stops.
If (.venv) still prefixes the shell prompt, the virtual environment remains active.
- Leave the active virtual environment by running this command:
deactivateWhat does this command do?
The deactivate command returns your shell to the system Python environment.
The .venv directory stays on disk for your next session.
- Check the beginning of your shell prompt.
You should no longer see (.venv) before the prompt.
Your Part 2 branches and all versioned assets remain available when you return.
Delete - I don't want to use this again
Deletion is irreversible. Your Part 1 repository stays intact because this cleanup targets the Part 2 branches and the disposable .venv directory.
- Switch back to the terminal running Uvicorn from earlier if the server is still active.
- Press Ctrl+C to stop the local server.
The server is stopped when your shell prompt returns.
If (.venv) appears before the prompt, leave the environment before deleting it.
- Leave the active virtual environment by running this command:
deactivateWhat does this command do?
The deactivate command returns your shell to the system Python environment.
This releases the active environment before you remove its directory.
- Check the beginning of your shell prompt.
You should no longer see (.venv) before the prompt.
- Use Git to switch from the local Part 2 branch to your Part 1 branch.
- Use Git to delete the local Part 2 branch.
- Check the local branch list in Git.
You should no longer see the Part 2 branch in the local branch list. Your Part 1 branch remains available.
- Return to the Part 1 repository on GitHub from earlier.
- Go to the repository branch management page.
- Locate the remote Part 2 branch.
- Use the branch deletion control to remove the remote Part 2 branch.
You should no longer see the Part 2 branch on the repository branch management page.
The leading dot can make .venv easy to miss in some file browsers.
- Return to the repository file tree from earlier.
- Enable hidden-file visibility if .venv does not appear.
- Delete the .venv directory.
- Refresh the repository file tree.
You should no longer see .venv in the repository. That's the cleanup complete: your Part 1 repository remains while the Part 2 branches and disposable environment are gone.
Nice Work!
Nice Work!
You did it! You built a client-ready AI triage service. Its deterministic guardrails now block unsafe regressions before merge.
You've learned how to:
- Built a 10-case deterministic baseline that protects priority decisions with frozen labels. Its ordered rules preserve routing outcomes plus human overrides.
- Created a two-tier evaluation suite that checks deterministic rules directly. Its model stub forces an unsafe downgrade. It also forces a provider failure. Human overrides bypass the model.
- Shipped POST /triage behind an enforced architecture boundary. The GitHub Actions gate runs one combined check for the architecture policy plus both evaluation tiers before merge.
- Completed the optional Secret Mission by triggering a forbidden SDK import in app/main.py. You repaired the provider boundary without changing behavior.
Ready to quiz yourself?