Decorators & Integration

Learn how to add OACP governance to your functions and integrate with LangGraph workflows.

@with_oacp Decorator

The main decorator for adding governance to any function. It handles logging, consensus, retries, and adaptive prompting.

Basic Usage

from oacp import with_oacp

@with_oacp(
    role="researcher",
    invariants=["factual_accuracy", "comprehensive_coverage"],
    log_inputs=True,
    log_outputs=True
)
def research_function(topic: str) -> dict:
    """Function with basic OACP governance."""
    # Your logic here
    return {"findings": "research results..."}

# Call the function
result = research_function("AI in healthcare")

Parameters

Parameter	Type	Description
`role`	str	Role identifier for this node
`invariants`	list[str]	List of invariants this node should maintain
`contract`	DecisionContract	Decision contract for voting requirements
`log_inputs`	bool	Whether to log function inputs (default: True)
`log_outputs`	bool	Whether to log function outputs (default: True)
`retry_policy`	RetryPolicy	Retry policy for consensus failures
`redact_keys`	list[str]	Keys to redact from logs
`adaptive_prompting`	bool	Enable adaptive prompting (default: True)

Functions with Consensus

Add decision contracts to require consensus before a function's output is accepted.

from oacp import with_oacp, decision_contract

@with_oacp(
    role="synthesizer",
    invariants=["comprehensive_integration", "actionable_conclusions"],
    contract=decision_contract(
        required_approvers=["researcher", "analyst", "critic", "synthesizer"],
        strategy="unanimous",
        timeout_seconds=30
    ),
    log_inputs=True,
    log_outputs=True
)
def synthesis_function(research_data, analysis_data, critique_data):
    """Function requiring unanimous consensus."""
    # Synthesis logic here
    final_report = create_synthesis(research_data, analysis_data, critique_data)
    
    # Voting happens automatically in background threads
    # Function returns only after consensus is achieved or timeout
    return final_report

wrap_node Function

Convenience function for adding OACP governance to existing functions without modifying them.

from oacp import wrap_node

def existing_function(data):
    """Existing function you want to add governance to."""
    return process_data(data)

# Wrap the function with OACP governance
governed_function = wrap_node(
    existing_function,
    role="processor",
    invariants=["data_integrity"],
    log_inputs=True,
    log_outputs=True
)

# Use in your workflow
result = governed_function(input_data)

LangGraph Integration

OACP integrates seamlessly with LangGraph workflows. Here's how to add governance to your graph nodes:

from langgraph.graph import StateGraph
from oacp import with_oacp, decision_contract
from typing import TypedDict

class WorkflowState(TypedDict):
    research_data: dict
    analysis_data: dict
    final_report: dict
    run_id: str

@with_oacp(
    role="researcher",
    invariants=["factual_accuracy"],
    log_inputs=True,
    log_outputs=True
)
def research_node(state: WorkflowState) -> dict:
    """Research node with governance."""
    # Research logic
    return {"research_data": research_results}

@with_oacp(
    role="synthesizer", 
    contract=decision_contract(
        required_approvers=["researcher", "analyst", "critic"],
        strategy="unanimous",
        timeout_seconds=30
    )
)
def synthesis_node(state: WorkflowState) -> dict:
    """Synthesis node requiring consensus."""
    # Synthesis logic requiring approval
    return {"final_report": synthesis_results}

# Build the graph
workflow = StateGraph(WorkflowState)
workflow.add_node("research", research_node)
workflow.add_node("synthesis", synthesis_node)
workflow.add_edge("research", "synthesis")

# Compile and run
app = workflow.compile()
result = app.invoke({"run_id": "example_run"})

Error Handling & Retries

OACP provides automatic retry mechanisms for consensus failures and other errors.

from oacp import with_oacp
from oacp.routing import RetryPolicy
from oacp.errors import OacpConsensusError

@with_oacp(
    role="analyzer",
    retry_policy=RetryPolicy(
        max_attempts=3,
        backoff_factor=2.0,
        max_backoff_seconds=60
    )
)
def analysis_with_retries(data):
    """Function with custom retry policy."""
    try:
        return analyze_data(data)
    except OacpConsensusError as e:
        # This will trigger a retry with adaptive prompting
        raise e

Accessing OACP Context

Access the current OACP context within your governed functions.

from oacp import with_oacp, current_context

@with_oacp(role="processor")
def context_aware_function(data):
    """Function that accesses OACP context."""
    ctx = current_context()
    
    print(f"Run ID: {ctx.run_id}")
    print(f"Node ID: {ctx.node_id}")
    print(f"Role: {ctx.role}")
    
    # Use context information in your logic
    if ctx.contract:
        print(f"Requires consensus from: {ctx.contract.required_approvers}")
    
    return process_with_context(data, ctx)