agentic-behavioral-contracts 0.1.0

Behavioral validation toolkit for AI agents in regulated environments. Checks whether agents follow the correct process — not just whether they produce correct outputs.

Agent Behavioral Validation for Regulated Environments

Status: Working — validated across 3 frameworks (Strands, LangChain, PydanticAI) + Bedrock AgentCore Goal: A practical toolkit for defining, validating, and auditing AI agent behavior in FinTech and HCLS.

---

The Problem

Teams in FinTech and HCLS are deploying AI agents for claims processing, clinical decision support, compliance review, and customer-facing copilots. These are high-stakes: wrong data, skipped steps, or hallucinated outputs can cause regulatory violations, financial loss, or patient harm.

Existing observability tools answer: "what happened?" Whether that's Langfuse, LangSmith, Arize Phoenix, AWS X-Ray, or raw OTEL traces — they all give you execution history. None of them answer: "did it do the right thing?"

Specifically, regulated teams cannot currently prove:

• The agent used the correct and current source data (not stale, not wrong patient/account)
• The agent followed the required reasoning path (didn't skip validation, didn't hallucinate policy)
• The output is consistent with expectations given the inputs
• When something fails, there is audit-ready evidence of exactly what happened and why

---

The Idea

A behavioral contract system for AI agents. Top-down, not bottom-up:

1. Define what correct behavior looks like — as executable contracts, not prose
2. Validate every execution against those contracts — in eval and in production
3. Surface violations and drift — before regulators or customers find them
4. Produce audit-ready evidence — not just traces, but verdicts with proof

---

Concrete Scenarios

1. Claims Processing Agent (HCLS)

An agent reviews insurance claims against formulary and coverage policies.

Behavioral contracts:

• MUST retrieve from the approved formulary database (not general knowledge)
• MUST check member coverage rules before rendering a decision
• MUST cite the specific policy section in approval or denial
• MUST escalate when claim type is outside trained categories
• MUST NOT use training data as a substitute for live formulary lookup

What a violation looks like:

• Agent approves a claim citing a policy section that doesn't exist
• Agent skips formulary lookup and answers from parametric knowledge
• Agent processes a claim type it wasn't designed for without escalating

2. KYC/AML Copilot (FinTech)

An agent assists analysts with know-your-customer and anti-money-laundering checks.

Behavioral contracts:

• MUST check against current sanctions lists (OFAC, EU, UN)
• MUST apply jurisdiction-specific rules based on customer location
• MUST flag high-risk indicators per regulatory thresholds
• MUST NOT auto-clear cases above risk threshold without human review
• MUST record which data sources were consulted for each determination

What a violation looks like:

• Agent clears a high-risk case without flagging for human review
• Agent applies US rules to an EU-jurisdiction customer
• Agent cites a sanctions list version that is 3 days stale

3. Clinical Decision Support (HCLS)

An agent assists clinicians with drug interaction checks and treatment recommendations.

Behavioral contracts:

• MUST retrieve from approved clinical guidelines (not general web)
• MUST check known drug interactions before recommending
• MUST include contraindication warnings when applicable
• MUST surface confidence level and flag low-confidence recommendations
• MUST NOT provide dosing recommendations outside approved ranges

What a violation looks like:

• Agent recommends a drug combination with a known interaction
• Agent provides a dosing recommendation from training data, not the formulary
• Agent gives a high-confidence answer on an edge case it shouldn't be confident about

4. Regulatory Compliance Review (FinTech)

An agent reviews transactions or documents for regulatory compliance.

Behavioral contracts:

• MUST reference current (not archived) regulations
• MUST flag when regulation version has changed since last review
• MUST escalate ambiguous cases rather than rendering a verdict
• MUST produce a citation trail linking conclusion to source regulation
• MUST NOT apply regulations from wrong jurisdiction

What a violation looks like:

• Agent applies a superseded regulation version
• Agent renders a confident verdict on an ambiguous case without escalation
• Agent cites a regulation that doesn't support its conclusion

---

Quick Start

# Install
pip install -e ".[dev]"

# Option 1: Auto-generate a contract from a golden trace (easiest)
agent-validate generate --traces my_trace.json --source-map map.json --name my-agent-v1
# Outputs YAML + Python contract automatically

# Option 2: Validate traces against a built-in contract
agent-validate run --contract claims-processing-v1 \
  --traces "traces/*.json" --source-map map.json --metadata claim_type=pharmacy

# Option 3: Use YAML contracts (no Python required)
agent-validate run --contract my-contract.yaml --traces "traces/*.json"

The 3-step workflow

1. Capture a golden trace — run your agent once on a known-good execution
2. Generate the contract — agent-validate generate --traces golden.json --output contract.yaml
3. Validate every execution — agent-validate run --contract contract.yaml --traces "traces/*.json" --fail-on-violation

No Python required. The generator analyzes what your agent did and creates constraints that require the same behavior on every future execution.

---

How It Works

Trace Adapter Architecture

The validator is trace-source-agnostic. Any observability tool or telemetry source is normalized into a common execution record before contracts are evaluated.

[OTEL collector]  ─┐
[Langfuse export]  ─┤
[LangSmith export] ─┼─→  Trace Adapter  ─→  Normalized Execution  ─→  Contract Validator  ─→  Verdict
[AWS X-Ray / CW]   ─┤
[Direct capture]   ─┘

See docs/trace-adapter-architecture.md for details.

MVP adapter: OpenTelemetry. OTEL is the first adapter because:

• Most agent frameworks already emit OTEL spans (Strands, LangChain, Bedrock)
• Langfuse, Arize Phoenix, and Braintrust can export to OTEL
• AWS native services (X-Ray, CloudWatch) support OTEL ingestion and export
• It's vendor-neutral — no coupling to a specific observability platform
• One adapter covers the widest surface area on day one

Behavioral Contracts as Code

from agent_validator import contract, source, step, output, escalation

@contract("claims-processing-v1")
class ClaimsProcessingContract:
    # Source constraints
    sources = source.must_retrieve_from(["formulary-db", "coverage-policy-api"])
    no_parametric = source.must_not_use_only_parametric_knowledge()

    # Step constraints
    required_steps = step.must_include(["formulary_lookup", "coverage_check"])
    step_order = step.must_precede("coverage_check", "render_decision")

    # Output constraints
    citations = output.must_contain_citations(min=1)
    decision_grounded = output.must_reference_source_for("decision")

    # Escalation constraints
    unknown_claim_type = escalation.when(
        condition="claim_type not in trained_categories",
        action="flag_for_human_review"
    )

Validation Against Executions

from agent_validator import validate
from agent_validator.adapters import OTELAdapter

# Normalize OTEL spans into an execution record
adapter = OTELAdapter()
execution = adapter.from_spans(spans)

# Validate against contract
result = validate(
    contract="claims-processing-v1",
    execution=execution,
)

# result.verdict: "pass" | "fail" | "warn"
# result.violations: [{ rule: "must_retrieve_from", detail: "formulary-db not consulted" }]
# result.evidence: [{ step: "coverage_check", sources_used: [...], citations: [...] }]
# result.audit_record: structured, exportable, timestamped

Continuous Production Monitoring

from agent_validator import monitor
from agent_validator.adapters import OTELAdapter

# Attach to OTEL collector — validates every execution as traces arrive
monitor(
    contract="claims-processing-v1",
    adapter=OTELAdapter(endpoint="http://localhost:4318"),
    on_violation="alert",     # or "block", "log", "escalate"
    drift_window="7d",        # detect behavioral drift over time
)

---

What This Is NOT

• NOT a tracing/observability tool (use Langfuse, LangSmith, OTEL, or AWS X-Ray for that)
• NOT a general-purpose eval framework (use DeepEval, RAGAS, or Braintrust for benchmarks)
• NOT a provenance platform (ideas 4+5 were heading there; this is sharper)
• NOT a policy engine (it validates agent behavior, not authorization)
• NOT a guardrails library (NeMo Guardrails, Guardrails AI filter I/O; this validates process)

It sits on top of existing observability and between eval frameworks and production monitoring.

It also complements evals — evals check if the output is correct, behavioral validation checks if it's correct for the right reasons. Together they catch the dangerous case where an agent gets the right answer by accident (skipped the formulary, guessed from training data). See docs/complementing-evals.md for the full analysis.

---

Competitive Landscape

See docs/competitive-landscape.md for detailed positioning.

Category	Tools	What they do	What they don't do
Guardrails	NeMo, Guardrails AI, Galileo Agent Control	Block bad I/O at runtime	Validate correct process or data sources
Observability	Langfuse, LangSmith, Arize, Openlayer	Show what happened	Judge if it was correct
Eval frameworks	DeepEval, RAGAS, Braintrust, Bloom	Test quality pre-deployment	Continuously validate in production
Behavioral contracts	Relari Agent Contracts, AgentAssert (research)	Generic behavioral validation	Domain-specific regulated scenarios, audit evidence
This project	—	Regulated behavioral validation with audit evidence	Not trying to replace any of the above

Key differentiators vs Relari Agent Contracts (closest competitor):

• Regulated-industry-specific contract templates, not generic
• Audit-ready evidence output, not just pass/fail
• OTEL-first adapter (works with existing observability), not its own instrumentation
• Data source validation as a core concern, not an afterthought

---

Why This Is Different From Ideas 4+5

See docs/differentiation-from-ideas-4-5.md

---

Current State

What's built and working:

1. Python library for defining behavioral contracts (6 constraint types, 3 domain templates)
2. OTEL trace adapter that normalizes OTLP JSON spans into execution records
3. Validator that checks contracts against normalized executions
4. Structured output — pass/fail/warn verdicts + evidence + audit records (JSON + terminal)
5. Three complete scenarios — claims processing, KYC/AML, clinical decision support
6. CLI — agent-validate with discover, dry-run, run (batch + CI mode), list-contracts
7. Discovery tools — describe_execution and dry_run for building contracts
8. Contract registry — built-in templates loadable by name, custom contracts from .py files
9. 48 tests passing
10. Validated against real traces from 3 frameworks — Strands, LangChain/LangGraph, PydanticAI
11. AgentCore integration — deployed agent, CloudWatch trace pull, end-to-end validation
12. Violation detection proven — catches agents that skip tools or fail to escalate
13. Audit persistence — save_audit_record(), save_batch_report(), save_to_s3()
14. Production monitoring — monitor_cloudwatch() and monitor_directory() with compliance rate tracking
15. All 3 domain templates tested — claims processing, KYC/AML, clinical decision support validated with real agents
16. Dynamic contracts — when() conditional constraints for process variants
17. Compliance tracking + drift detection — ComplianceTracker with sliding windows, per-constraint rates, threshold alerts
18. Auto-generate contracts — agent-validate generate creates contracts from golden traces (YAML + Python output)
19. YAML contracts — define and load contracts without writing Python (67 tests passing)

Real-world validation results

Tested against real agents calling Claude on Bedrock across three frameworks:

Framework	Environment	Tool Spans	Result
Strands	Local + Bedrock AgentCore	execute_tool + gen_ai.tool.name	6/6 PASS
LangChain/LangGraph	Local	execute_tool + gen_ai.tool.name	6/6 PASS
PydanticAI	Local	running tool + gen_ai.tool.name	6/6 PASS

Violation detection:

Scenario	Result
Agent answers from parametric knowledge (no tools)	5/6 FAIL — caught
Agent doesn't escalate unknown claim type	6/6 FAIL — caught

Same adapter, same contract, same tool_source_map — works across all three frameworks. See examples-aws/ for the full end-to-end tests.

What to build next:

• OTEL Collector plugin for inline validation (see packaging doc)
• Drift detection over time (compliance rate trends)
• UI or dashboard
• Additional frameworks (CrewAI has Bedrock tool-use compatibility issues)

---

Roadmap: From Library to Turnkey

What exists today

A Python library and a CLI. Both are functional.

CLI (no Python knowledge required):

# List available contracts
agent-validate list-contracts

# Inspect a trace — see step names, data sources, structure
agent-validate discover --traces traces/sample.json --source-map map.json

# Test a contract against traces — detailed output per constraint
agent-validate dry-run --contract claims-processing-v1 --traces traces/sample.json \
  --source-map map.json --metadata claim_type=pharmacy

# Validate a batch of traces with JSON export and CI exit codes
agent-validate run --contract claims-processing-v1 --traces "traces/*.json" \
  --source-map map.json --quiet --output report.json --fail-on-violation

Library (for integration into Python code):

from agent_validator import validate, create_audit_record
from agent_validator.adapters.otel import OTELAdapter

execution = adapter.normalize(otel_spans)
verdict = validate(contract, execution)
record = create_audit_record(verdict, execution, contract.version)

Who uses what

Who	Tool	How
Agent developer	Library	In tests and scripts
Platform/MLOps team	CLI	In CI pipelines with --fail-on-violation
Compliance officer	CLI	Batch runs with --output report.json
External auditor	CLI output	Reviews JSON audit records

What's next: packaging tiers beyond the CLI

Tier 2: OTEL Collector plugin — inline production validation

# otel-collector-config.yaml
processors:
  agent_validator:
    contract: claims-processing-v1
    tool_source_map:
      lookup_formulary: { source_name: formulary-db, source_type: database }
      check_coverage: { source_name: coverage-policy, source_type: api }
    on_violation: alert
    audit_output: s3://my-bucket/audit-records/

Validates every trace as it flows through infrastructure the customer already runs. Zero code changes to the agent or pipeline. The most "set it and forget it" option. Build this when there's demand from teams already running OTEL collectors.

Tier 3: Cloud-native deployment — managed validation

Agent → OTEL Collector → S3 (traces)
                              ↓
                     Lambda (validates) → DynamoDB (audit records)
                              ↓                      ↓
                     SNS (violation alerts)    Dashboard (compliance view)

Scales automatically, audit records are durable by default, compliance teams get a dashboard instead of JSON files. This is a product, not a library — build it only if the library and CLI prove the model works and there's clear demand.

Decision criteria for each tier

Tier	Status
Library	Done — validated against 3 frameworks (Strands, LangChain, PydanticAI), locally and on AgentCore
CLI	Done — discover, dry-run, run, list-contracts all functional
Audit persistence	Done — save_audit_record(), save_batch_report(), save_to_s3() in store.py
Multi-framework	Done — same adapter works across Strands, LangChain/LangGraph, PydanticAI
Violation detection	Done — catches parametric-only answers and missing escalations
Production monitoring	Done — monitor_cloudwatch() and monitor_directory() with compliance rate
OTEL Collector plugin	Next — when teams want inline production validation without code changes
Cloud-native / managed	When there's demand for a product, not just a tool

The library, CLI, audit persistence, and multi-framework support are all proven. The next step is the OTEL Collector plugin for inline production validation.

---

Open Questions

Resolved

• How do you define "must retrieve from X" when the agent uses tool calls that abstract the source? Resolved: The OTEL adapter's tool_source_map bridges tool names to data source identities. You configure "lookup_formulary" → {"source_name": "formulary-db", "source_type": "database"} and the adapter maps tool calls to data sources during normalization. See adapters/otel.py.

• Can citation checking be done from OTEL spans alone, or does it require an LLM-as-judge step? Partially resolved: Citation presence can be checked from spans using regex heuristics — the OutputMustContainCitations constraint matches patterns like "Section 4.2.1", "[1]", etc. Citation accuracy (does the cited section actually support the conclusion?) still requires LLM-as-judge. This is a known fidelity gap: the CheckMethod.HEURISTIC vs CheckMethod.LLM_JUDGE distinction in the model already accounts for this. An LLM-judge constraint is a natural next step but not required for MVP.

• What does a useful audit record look like for a FinTech compliance team vs an HCLS audit team? Partially resolved: The AuditRecord structure is defined — verdict + execution summary + evidence package, serializable to JSON. See audit.py. Still open: whether FinTech and HCLS teams need different audit output formats, fields, or evidence depth. Needs validation with real compliance personas before assuming the current shape is sufficient.

Resolved (from real-world testing)

• Are OTEL GenAI semantic conventions stable enough to build on? Resolved: Yes. Tested against 3 frameworks (Strands, LangChain/LangGraph, PydanticAI). The adapter relies on gen_ai.tool.name for tool identification — reliably emitted by all three. Message content follows OTEL GenAI spec (span events or span attributes depending on framework). The adapter reads from both. This is framework-agnostic.

• How rich are Strands Agents' OTEL spans? Resolved: Rich enough for all 6 constraint types. See examples-aws/.

• Does the adapter work across frameworks? Resolved: Yes. Same adapter, same contract, same tool_source_map works across:

  • Strands (12 spans, native OTEL, execute_tool span names)
  • LangChain/LangGraph (32 spans, opentelemetry-instrumentation-langchain, execute_tool span names)
  • PydanticAI (7 spans, built-in instrument=InstrumentationSettings(tracer_provider=...), running tool span names) All three emit gen_ai.tool.name on tool spans. The adapter classifies by this attribute.

• Audit record persistence. Resolved: save_audit_record(), save_batch_report(), and save_to_s3() in src/agent_validator/store.py. Individual JSON files, batch reports with summaries, and S3 with date-partitioned keys.

Resolved (dynamic contracts)

• Should contracts be static or dynamic? Resolved: Added when() conditional constraints. One contract handles multiple process variants — constraints activate based on execution metadata.

  Contract(constraints=[
      must_not_use_only_parametric_knowledge(),          # always
      when({"claim_type": "pharmacy"}, must_include_steps(["lookup_formulary"])),
      when({"claim_type": "dental"}, must_include_steps(["verify_dental_benefits"])),
      when(lambda ex: ex.metadata.get("risk_score", 0) >= 70, must_include_steps(["enhanced_review"])),
  ])
  

  Supports dict conditions (metadata key-value matching), callable conditions, and multi-key conditions (all must match). Skipped constraints count as PASS. 9 tests covering all patterns.

Resolved (probabilistic behavior)

• How do you handle probabilistic behavior? Resolved: Per-execution validation stays binary (pass/fail). A new ComplianceTracker aggregates verdicts over a sliding window and detects drift:

  from agent_validator import ComplianceTracker, CompliancePolicy
  
  tracker = ComplianceTracker(
      policy=CompliancePolicy(
          overall_threshold=0.95,        # 95% overall pass rate required
          min_window_size=10,            # don't alert until 10 executions seen
          constraint_thresholds={        # per-constraint overrides
              "must_include_steps(...)": 0.99,
          },
      ),
      window_size=100,                   # sliding window of last 100 executions
      on_alert=lambda alert: send_to_slack(alert),
  )
  
  # Feed verdicts as they arrive
  alerts = tracker.record(verdict)       # returns DriftAlert list
  print(tracker.overall_compliance)      # 0.95
  print(tracker.summary())              # per-constraint rates + failing constraints
  

  10 tests covering: empty state, mixed results, drift detection, no-alert-below-minimum, per-constraint thresholds, sliding window recovery, summary structure.

All Open Questions Resolved

No remaining open questions from the original design.