Add deterministic safety certification to LangGraph agent workflows. This example demonstrates how to integrate the QAE safety kernel into a multi-step agent pipeline, certifying tool calls before execution and blocking/escalating unsafe actions.
Agent State Graph
↓
Propose Tool Call (LLM)
↓
QAE Certification Node
├─→ Extract state deltas from tool call
├─→ Run through SafetyCertifier
└─→ Decision: Certified / CertifiedWithWarning / EscalateToHuman / Blocked
↓
Execute Tool (or pause/reject)
↓
Observe → Loop or Complete
- Install dependencies:
pip install -r requirements.txt- Run the basic certified agent example:
python -m src.certified_agentExpected output: Agent proposes a tool call, QAE certifies it, and displays the certificate decision.
- Run the custom domain adapter example:
python -m src.custom_domain_exampleDemonstrates how to create domain-specific constraint channels.
src/certified_agent.py— LangGraph state graph with QAE certification nodesrc/custom_domain_example.py— Custom DomainAdapter and ConstraintChannel subclassesrequirements.txt— Python dependencies
Stateful certifier wrapping a DomainAdapter:
from qae_safety import AgenticAdapter, SafetyCertifier, SimpleAction, StateDelta
adapter = AgenticAdapter(budget_limit=100.0, rate_limit=50.0)
certifier = SafetyCertifier(adapter)
action = SimpleAction(
action_id="tool-call-1",
agent_id="agent-1",
state_deltas=[
StateDelta(dimension="scope_score", from_value=1.0, to_value=0.8),
]
)
cert = certifier.certify(action)
print(f"Decision: {cert.decision}")
print(f"Zone: {cert.zone}")
print(f"Margins: {cert.margins}")| Margin Range | Zone | Decision | Action |
|---|---|---|---|
| > 0.6 | Safe | Certified | Execute immediately |
| (0.3, 0.6] | Caution | CertifiedWithWarning | Execute with logs/warnings |
| (0.1, 0.3] | Danger | EscalateToHuman | Pause for human review |
| ≤ 0.1 | Danger | Blocked | Reject |
Extend the safety kernel for your domain:
from qae_safety import DomainAdapter, ConstraintChannel
class MyConstraint(ConstraintChannel):
def name(self):
return "my_constraint"
def evaluate(self, state):
# Return margin in [0, 1]
return 0.75
def dimension_names(self):
return ["my_dimension"]
class MyAdapter(DomainAdapter):
def domain_name(self):
return "my_domain"
def constraint_channels(self):
return [MyConstraint()]
def map_action_to_state(self, action):
return [1.0]
def detect_regime_change(self, current, proposed):
return FalseFor the theory behind QAE, see:
- Paper: "Deterministic Portfolio Risk Certification" (DOI: 10.6084/m9.figshare.31742857)
- Website: qaesubstrate.com
This example is provided as part of the QAE project. See LICENSE in the root repository.