Metadata-Version: 2.4
Name: agentic-behavioral-contracts
Version: 0.1.0
Summary: Behavioral validation toolkit for AI agents in regulated environments. Checks whether agents follow the correct process — not just whether they produce correct outputs.
Project-URL: Homepage, https://github.com/MikeBrouwers/agentic-behavioral-contracts
Project-URL: Repository, https://github.com/MikeBrouwers/agentic-behavioral-contracts
Project-URL: Issues, https://github.com/MikeBrouwers/agentic-behavioral-contracts/issues
License: Apache-2.0
License-File: LICENSE
Keywords: agents,ai,behavioral-validation,compliance,fintech,healthcare,opentelemetry,otel,regulated
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Quality Assurance
Classifier: Topic :: Software Development :: Testing
Requires-Python: >=3.9
Requires-Dist: opentelemetry-api>=1.20.0
Requires-Dist: opentelemetry-sdk>=1.20.0
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Provides-Extra: examples
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.20.0; extra == 'examples'
Requires-Dist: python-dotenv>=1.0.0; extra == 'examples'
Requires-Dist: strands-agents-bedrock>=0.1.0; extra == 'examples'
Requires-Dist: strands-agents>=0.1.0; extra == 'examples'
Provides-Extra: yaml
Requires-Dist: pyyaml>=6.0; extra == 'yaml'
Description-Content-Type: text/markdown

# 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

```bash
# 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](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

```python
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

```python
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

```python
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](docs/complementing-evals.md) for the full analysis.

---

## Competitive Landscape

See [docs/competitive-landscape.md](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](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/](examples-aws/) for the full end-to-end tests.

What to build next:

- OTEL Collector plugin for inline validation (see [packaging doc](docs/packaging-cli-and-collector.md))
- 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):

```bash
# 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):

```python
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

```yaml
# 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/](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.

  ```python
  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:

  ```python
  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.
