Agents repeat tool calls, retry failing dependencies, and bounce work between specialists. BehaviorWeave detects these patterns deterministically and returns typed interventions: your application decides when to nudge, escalate, request review, or stop.

Highlights

• Deterministic and explainable: pattern, count, threshold, policy, and provenance; no model call is needed to evaluate a policy.
• Async-first: engine, storage, audit, and lifecycle operations are awaitable.
• LangChain middleware: guard create_agent and Deep Agents tool calls; scope by thread, append guidance, block configured kinds, and track outcomes.
• Customizable: ABCs for patterns, stores, and sinks; inject every agxai capability, plugin, configuration, and exporter.
• Durable history: LangGraph BaseStore adapter and PostgreSQL/Redis reference stores; persisted middleware halts and reusable conformance tests.
• Reliable: scope isolation, idempotent redelivery, atomic state transitions, and 100% source line and branch coverage.

Install

pip install behaviorweave

Or uv add behaviorweave. Supports Python 3.12 through 3.15.

Quickstart

import asyncio
from behaviorweave import BehaviorEngine, BehaviorEvent, InterventionType, PolicyRule


async def main():
    async with BehaviorEngine(
        policies=[
            PolicyRule("nudge", "repeated_tool_call", 2, InterventionType.NUDGE),
            PolicyRule("stop", "repeated_tool_call", 3, InterventionType.STOP),
        ]
    ) as engine:
        kinds = []
        for _ in range(3):
            decision = await engine.process(
                BehaviorEvent.tool_call("get_alarm", {"machine": "ETCH-3"}, scope="incident-42")
            )
            kinds.append(decision.intervention.kind.value)
        print(kinds)  # ['noop', 'nudge', 'stop']


asyncio.run(main())

Guard a LangChain agent

from langchain.agents import create_agent
from behaviorweave.integrations.langchain import BehaviorWeaveMiddleware

agent = create_agent(model, tools=[get_alarm], middleware=[BehaviorWeaveMiddleware(engine)])
result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Inspect ETCH-3."}]},
    config={"configurable": {"thread_id": "incident-42"}},
)

The second identical call receives [BehaviorWeave:nudge] guidance. The third is blocked before the tool runs. The engine returns decisions; enforcement belongs to the middleware or your host application.

Built-in patterns

Policies select thresholds, priorities, per-policy cooldowns, once-only behavior, and intervention kinds. Configure the built-in pattern classes or subclass the Pattern ABC.

Integrations and customization

See the downstream inventory. agxai is used for provenance and explanations; BehaviorWeave owns behavioral decisions.

Examples and Agent Skills

The twenty async examples cover middleware, MCP, graph loops, Swarm, Deep Agents, concurrent runs, durable databases, agxai semantic audit, OpenTelemetry traces and metrics, and batch replay. Configure model credentials through environment variables; database/OTel examples support local Podman services.

The Agent Skills teach agents how to integrate and extend the package, with executable checks and evaluations.

Documentation and project

• Quickstart
• Concepts
• Integrations
• API reference
• Changelog
• Contributing
• Code of Conduct
• Security

Apache License 2.0. See LICENSE.