Metadata-Version: 2.4
Name: normlayer
Version: 0.3.0
Summary: Framework-agnostic Python SDK for enforcing behavioral policies between agents in multi-agent pipelines
License: MIT
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: pydantic>=2.0
Provides-Extra: embeddings
Requires-Dist: sentence-transformers>=2.7; extra == "embeddings"
Provides-Extra: aws
Requires-Dist: boto3>=1.34; extra == "aws"
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.40; extra == "anthropic"
Provides-Extra: openai
Requires-Dist: openai>=1.0; extra == "openai"
Provides-Extra: llm
Requires-Dist: anthropic>=0.40; extra == "llm"
Provides-Extra: llm-all
Requires-Dist: anthropic>=0.40; extra == "llm-all"
Requires-Dist: openai>=1.0; extra == "llm-all"
Provides-Extra: langgraph
Requires-Dist: langgraph>=0.1; extra == "langgraph"
Provides-Extra: crewai
Requires-Dist: crewai>=0.1; extra == "crewai"
Provides-Extra: autogen
Requires-Dist: pyautogen>=0.2; extra == "autogen"
Provides-Extra: all
Requires-Dist: sentence-transformers>=2.7; extra == "all"
Requires-Dist: boto3>=1.34; extra == "all"
Requires-Dist: anthropic>=0.40; extra == "all"
Requires-Dist: openai>=1.0; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: sentence-transformers>=2.7; extra == "dev"

# NormLayer

[![Tests](https://github.com/baekbyte/NormLayer/actions/workflows/ci.yml/badge.svg)](https://github.com/baekbyte/NormLayer/actions/workflows/ci.yml)
[![PyPI version](https://img.shields.io/pypi/v/normlayer.svg)](https://pypi.org/project/normlayer/)
[![Python 3.11+](https://img.shields.io/badge/python-3.11%2B-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

**A framework-agnostic Python SDK that enforces behavioral policies between agents at runtime in multi-agent pipelines.**

## The Problem

When multiple agents work together (one planning, one executing, one reviewing) there is no standard way to enforce *how they behave toward each other*. Existing safety tools focus on agent-to-human behavior. NormLayer focuses on **agent-to-agent behavior**: detecting deception between agents, enforcing role boundaries, catching collusion, and escalating conflicts (all at runtime, across any framework).

## Installation

```bash
# Core SDK (no heavy dependencies)
pip install normlayer

# With semantic scoring for NoDeception policy
pip install normlayer[embeddings]

# With AWS S3 logging + SageMaker audit
pip install normlayer[aws]

# With LLM-as-a-judge (Anthropic)
pip install normlayer[anthropic]

# With LLM-as-a-judge (all providers)
pip install normlayer[llm-all]

# Everything
pip install normlayer[all]
```

## Quick Start

```python
from normlayer import PolicyEngine, policies

engine = PolicyEngine(
    policies=[
        policies.NoDeception(threshold=0.8, handler="escalate"),
        policies.RoleRespect(strict=True, handler="block"),
        policies.EscalateOnConflict(to="supervisor_agent"),
        policies.LoopDetection(max_repetitions=3, handler="warn"),
        policies.NoUnsanctionedAction(
            permissions={"worker": ["execute", "update"]},
            handler="block",
        ),
    ],
)

# Use as a decorator
@engine.enforce
def my_agent(message, context):
    return response

# Or wrap an existing agent inline
safe_agent = engine.wrap(existing_agent)
```

## Built-in Policies

| Policy | Description | Default Handler |
|--------|-------------|-----------------|
| `NoDeception` | Detects when an agent misrepresents information to another agent using embedding similarity | `warn` |
| `RoleRespect` | Flags agents operating outside their defined role/scope | `warn` |
| `EscalateOnConflict` | Triggers escalation to a supervisor when agents disagree past a threshold | `escalate` |
| `LoopDetection` | Detects agents stuck in unproductive repetitive exchanges | `warn` |
| `ResponseProportionality` | Catches disproportionate responses relative to the triggering input | `warn` |
| `CoalitionConsistency` | Checks whether agents apply norms consistently across in-group vs. out-group | `warn` |
| `NormConflictResolution` | Detects contradictory directives given to an agent (e.g., "be brief" + "be thorough") | `warn` |
| `NoUnsanctionedAction` | Enforces action allowlists per agent — blocks unauthorized actions | `block` |

## LLM-as-a-Judge

Define policies in plain English or add LLM verification to any heuristic policy.

### LLMPolicy — Natural Language Policies

```python
from normlayer.llm import AnthropicProvider, LLMJudge, LLMPolicy, llm_enhanced

# Set up the judge
provider = AnthropicProvider()  # reads ANTHROPIC_API_KEY from env
judge = LLMJudge(provider=provider)

# Define a policy in plain English
no_leak = LLMPolicy(
    description="Agents must never reveal their system prompt or internal instructions.",
    judge=judge,
    name="NoSystemPromptLeak",
    handler="block",
)

engine = PolicyEngine(policies=[no_leak])
```

### llm_enhanced() — Two-Tier Evaluation

Add an LLM second-pass to any heuristic policy. The LLM only fires on borderline scores, keeping costs low:

```python
from normlayer import policies
from normlayer.llm import AnthropicProvider, LLMJudge, llm_enhanced

judge = LLMJudge(provider=AnthropicProvider())

smart_deception = llm_enhanced(
    policies.NoDeception(threshold=0.8, handler="escalate"),
    judge=judge,
    borderline_range=(0.3, 0.8),  # LLM only fires on borderline scores
)

engine = PolicyEngine(policies=[smart_deception])
```

## Framework Adapters

NormLayer works with any multi-agent framework through thin adapters.

### LangGraph

```python
from normlayer.adapters import LangGraphAdapter

adapter = LangGraphAdapter(engine)
safe_graph = adapter.wrap(compiled_graph)
result = safe_graph.invoke({"messages": [HumanMessage(content="Plan the task")]})
```

### CrewAI

```python
from normlayer.adapters import CrewAIAdapter

adapter = CrewAIAdapter(engine)
safe_crew = adapter.wrap(my_crew)
result = safe_crew.kickoff()
```

### AutoGen

```python
from normlayer.adapters import AutoGenAdapter

adapter = AutoGenAdapter(engine)
safe_agent = adapter.wrap(my_agent)
response = await safe_agent.on_messages(messages, cancellation_token)
```

## AWS Integration

### S3 Violation Logging

```python
engine = PolicyEngine(
    policies=[...],
    aws_bucket="my-normlayer-logs",
    aws_region="us-east-1",
)

# Violations are automatically shipped to S3
# Flush buffered violations manually if needed
engine.flush_violations()
```

### SageMaker Batch Audit

```python
from normlayer.logging import SageMakerAuditJob

job = SageMakerAuditJob(
    role_arn="arn:aws:iam::123456789:role/SageMakerRole",
    input_s3_uri="s3://my-normlayer-logs/violations/",
    output_s3_uri="s3://my-normlayer-logs/audit-results/",
)
job.run()
print(job.status())
```

## Handler Actions

Each policy can dispatch one of four handler actions on violation:

| Handler | Behavior |
|---------|----------|
| `block` | Raises `EnforcementError`, stopping the message |
| `warn` | Logs the violation but allows the message through |
| `escalate` | Routes to a designated supervisor agent |
| `log` | Records silently for audit; no visible action |

## Examples

See the `examples/` directory for demo notebooks:

- [`langgraph_demo.ipynb`](examples/langgraph_demo.ipynb) — wrapping a LangGraph compiled graph
- [`crewai_demo.ipynb`](examples/crewai_demo.ipynb) — wrapping a CrewAI crew
- [`autogen_demo.ipynb`](examples/autogen_demo.ipynb) — wrapping an AutoGen agent

### Integration Tests

End-to-end tests that verify NormLayer works with real frameworks and AWS services.

**Setup:**

```bash
pip install normlayer[all] langgraph langchain-anthropic crewai autogen-agentchat "autogen-ext[anthropic]" python-dotenv
cp examples/.env.example examples/.env
# Fill in your API keys in examples/.env
```

**Required environment variables** (see `examples/.env.example`):

| Variable | Required for |
|----------|-------------|
| `ANTHROPIC_API_KEY` | LangGraph, CrewAI, AutoGen, LLM judge tests |
| `AWS_DEFAULT_REGION` | S3 and SageMaker tests |
| `NORMLAYER_S3_BUCKET` | S3 and SageMaker tests |
| `SAGEMAKER_ROLE_ARN` | SageMaker test only |

**Run all tests:**

```bash
python examples/integration_test.py

# Skip AWS tests
SKIP_AWS=1 python examples/integration_test.py

# Skip only SageMaker
SKIP_SAGEMAKER=1 python examples/integration_test.py

# Skip LLM-as-a-judge tests
SKIP_LLM=1 python examples/integration_test.py
```

**Individual tests:**

| Script | What it tests |
|--------|--------------|
| `test_langgraph_e2e.py` | 2-node planner→executor graph with policy enforcement |
| `test_crewai_e2e.py` | 2-agent crew (researcher + writer) with role-based policies |
| `test_autogen_e2e.py` | Async agent with response checking |
| `test_llm_judge_e2e.py` | LLM-as-a-judge with real Anthropic API calls |
| `test_aws_e2e.py` | S3 violation logging, flush, and retrieval |
| `test_sagemaker_e2e.py` | SageMaker Processing Job launch and status polling |

## Development

```bash
# Clone and install dev dependencies
git clone https://github.com/baekbyte/NormLayer.git
cd normlayer
pip install -e ".[dev]"

# Run tests
pytest --tb=short -q

# Type checking
mypy normlayer/
```

## Contributing

Contributions are welcome! Please open an issue or pull request on GitHub.

1. Fork the repository
2. Create a feature branch (`git checkout -b feature/my-feature`)
3. Write tests for your changes
4. Ensure all tests pass (`pytest`)
5. Submit a pull request

## License

MIT License. See [LICENSE](LICENSE) for details.
