Enterprise AI Agents: Traceability, Atomicity, and Memory Persistence with AgentHelm

When deploying AI agents in enterprise environments, three requirements typically surface: every action must be traceable, multi-step operations must be atomic, and context must persist across sessions. AgentHelm v0.3.0 addresses all three.

This walkthrough demonstrates building a contract processing system with two specialized agents and full observability.

The Enterprise Requirements

Before diving into code, let's establish what enterprise deployments demand:

1. Traceability: Every tool call must be logged with inputs, outputs, timing, and cost
2. Atomicity: If step 3 of 5 fails, steps 1-2 must be rolled back
3. Memory Persistence: Agent context survives restarts and can be audited
4. Cost Visibility: Know exactly what each operation costs before the invoice arrives

Architecture Overview

┌────────────────────────────────────────────────────────────┐
│                      PlannerAgent                          │
│              (Generates execution blueprint)               │
└────────────────────────────────────────────────────────────┘
                              │
                              ▼
              ┌───────────────┴───────────────┐
              │         Orchestrator          │
              │   (Saga pattern execution)    │
              └───────────────┬───────────────┘
                              │
              ┌───────────────┴───────────────┐
              ▼                               ▼
┌─────────────────────┐         ┌─────────────────────┐
│   AnalysisAgent     │         │   DocumentAgent     │
│ (Extract & analyze) │         │  (Generate & save)  │
└─────────────────────┘         └─────────────────────┘
              │                               │
              └───────────────┬───────────────┘
                              ▼
              ┌───────────────────────────────┐
              │          MemoryHub            │
              │  (SQLite short-term + Qdrant) │
              └───────────────────────────────┘
                              │
                              ▼
              ┌───────────────────────────────┐
              │       ExecutionTracer         │
              │  (SQLite + OpenTelemetry)     │
              └───────────────────────────────┘
                              │
                              ▼
              ┌───────────────────────────────┐
              │           Jaeger              │
              │   (Trace visualization)       │
              └───────────────────────────────┘

Setup

pip install agenthelm

import dspy
from agenthelm import (
    ToolAgent, PlannerAgent, Orchestrator, AgentRegistry,
    tool, MemoryHub, ExecutionTracer
)
from agenthelm.core.storage import SqliteStorage
from agenthelm.tracing import init_tracing

lm = dspy.LM("mistral/mistral-large-latest")

Configure Observability First

In enterprise deployments, observability is not an afterthought.

# Initialize OpenTelemetry with Jaeger
init_tracing(
    service_name="contract-processor",
    otlp_endpoint="http://jaeger:4317",
    enabled=True
)

# Create execution tracer with persistent storage
tracer = ExecutionTracer(
    storage=SqliteStorage("/var/log/agenthelm/traces.db"),
    session_id="contract-batch-2025-12-26"
)

Every tool execution is now:

• Logged to SQLite with full inputs/outputs
• Exported to Jaeger for distributed tracing
• Tagged with session ID for batch correlation

Configure Memory Persistence

# Production memory configuration
memory = MemoryHub(
    data_dir="/var/data/agenthelm",  # Local persistence
    # Or for network mode:
    # redis_url="redis://cache.internal:6379",
    # qdrant_url="http://qdrant.internal:6333"
)

MemoryHub provides:

• Short-term memory: Session state with TTL (SQLite locally, Redis for distributed)
• Semantic memory: Vector search for context retrieval (Qdrant with FastEmbed)

Define Tools with Compensation

Atomicity requires compensating actions for rollback.

@tool()
def extract_contract_data(document_path: str) -> dict:
    """Extract structured data from a contract document."""
    # Simulated extraction
    return {
        "parties": ["Acme Corp", "Widget Inc"],
        "value": 150000,
        "terms": "12 months",
        "effective_date": "2025-01-01"
    }

@tool()
def validate_compliance(contract_data: dict) -> dict:
    """Validate contract against compliance rules."""
    issues = []
    if contract_data.get("value", 0) > 100000:
        issues.append("Requires senior approval for contracts > $100k")
    return {"valid": len(issues) == 0, "issues": issues}

@tool(compensating_tool="delete_record")
def create_record(contract_data: dict, record_type: str) -> str:
    """Create a record in the system."""
    record_id = f"REC-{hash(str(contract_data)) % 10000:04d}"
    # In production: database insert
    return record_id

@tool()
def delete_record(record_id: str) -> str:
    """Delete a record (compensation action)."""
    # In production: database delete
    return f"Deleted {record_id}"

@tool(compensating_tool="archive_document")
def generate_summary(contract_data: dict, output_path: str) -> str:
    """Generate and save a contract summary document."""
    content = f"Contract Summary: {contract_data}"
    with open(output_path, "w") as f:
        f.write(content)
    return output_path

@tool()
def archive_document(output_path: str) -> str:
    """Archive a document (compensation action)."""
    import os
    if os.path.exists(output_path):
        os.rename(output_path, f"{output_path}.archived")
    return f"Archived {output_path}"

Note the compensating_tool parameter. When the Orchestrator detects a failure, it automatically calls these in reverse order.

Create Specialized Agents

# Agent 1: Analysis specialist
analysis_agent = ToolAgent(
    name="analyst",
    lm=lm,
    tools=[extract_contract_data, validate_compliance],
    tracer=tracer,
    memory=memory,
    role="You are a contract analysis specialist. Extract data accurately."
)

# Agent 2: Document specialist  
document_agent = ToolAgent(
    name="documenter",
    lm=lm,
    tools=[create_record, generate_summary],
    tracer=tracer,
    memory=memory,
    role="You are a document management specialist. Create records precisely."
)

# Register agents
registry = AgentRegistry()
registry.register(analysis_agent)
registry.register(document_agent)

Both agents share:

• The same tracer (unified trace storage)
• The same memory hub (shared context)

Generate the Execution Plan

# Planner agent has visibility into all tools
planner = PlannerAgent(
    name="planner",
    lm=lm,
    tools=[
        extract_contract_data, validate_compliance,
        create_record, generate_summary
    ]
)

plan = planner.plan(
    "Process contract.pdf: extract data, validate compliance, "
    "create a system record, and generate a summary document"
)

print(plan.to_yaml())

Generated plan:

goal: Process contract and create records
reasoning: |
  Sequential process: extract first, then validate, then create
  record, finally generate summary. Each step depends on previous.
steps:
  - id: extract
    agent: analyst
    tool: extract_contract_data
    description: Extract structured data from contract
    args:
      document_path: "contract.pdf"

  - id: validate
    agent: analyst
    tool: validate_compliance
    description: Check against compliance rules
    depends_on: [extract]
    args:
      contract_data: "${extract.result}"

  - id: record
    agent: documenter
    tool: create_record
    description: Create system record
    depends_on: [validate]
    args:
      contract_data: "${extract.result}"
      record_type: "contract"

  - id: summary
    agent: documenter
    tool: generate_summary
    description: Generate summary document
    depends_on: [record]
    args:
      contract_data: "${extract.result}"
      output_path: "/output/contract_summary.md"

Review and Edit the Plan

Before execution, the plan can be reviewed and modified.

# Save plan for review
plan_path = "/reviews/contract_plan.yaml"
with open(plan_path, "w") as f:
    f.write(plan.to_yaml())

# Manual review happens here...
# Reviewer can edit the YAML, add steps, modify args

# Load reviewed plan
from agenthelm import Plan
reviewed_plan = Plan.from_yaml(open(plan_path).read())

# Approve for execution
reviewed_plan.approved = True

In production, this review step integrates with your approval workflow; Slack notifications, PR-based reviews, or manual sign-off.

Execute with Saga Rollback

orchestrator = Orchestrator(
    registry=registry,
    enable_rollback=True  # Saga pattern enabled
)

result = await orchestrator.execute(reviewed_plan)

If generate_summary fails after create_record succeeds:

1. generate_summary marked as FAILED
2. Orchestrator triggers rollback
3. delete_record called automatically (compensating action for create_record)
4. System returns to consistent state

Inspect Traces

After execution, full traceability:

# Programmatic access
for event in result.events:
    print(f"{event.tool_name}: {event.execution_time:.3f}s, ${event.estimated_cost_usd:.4f}")

# Summary
print(f"Total cost: ${result.total_cost_usd:.4f}")
print(f"Total tokens: {result.token_usage.total_tokens}")

Via CLI:

# List recent traces
agenthelm traces list -s /var/log/agenthelm/traces.db

# Filter by tool
agenthelm traces filter --tool create_record --status success

# Export for audit
agenthelm traces export -o audit_report.json -f json
agenthelm traces export -o audit_report.csv -f csv

Memory for Context Continuity

Store and retrieve context across sessions:

from agenthelm.memory import MemoryContext

async with MemoryContext(memory, session_id="contract-batch-2025-12-26") as ctx:
    # Store processing context
    await ctx.set("last_processed_contract", "contract.pdf")
    await ctx.set("batch_status", {"processed": 1, "failed": 0})

    # Store semantic memory for future retrieval
    await ctx.store_memory(
        "Contract with Acme Corp processed successfully. Value: $150k, 12 months.",
        metadata={"contract_id": "contract.pdf", "status": "complete"}
    )

# Later, in another session
async with MemoryContext(memory, session_id="new-session") as ctx:
    # Recall relevant past contracts
    results = await ctx.recall("Acme Corp contracts", top_k=5)
    for r in results:
        print(f"Score: {r.score:.2f} - {r.text}")

Jaeger Integration

With OpenTelemetry configured, view traces in Jaeger:

# Start Jaeger
docker run -d -p 16686:16686 -p 4317:4317 jaegertracing/all-in-one

# Run your agent workflow
python process_contracts.py

# Open Jaeger UI
open http://localhost:16686

Each agent execution appears as a span with:

• Tool name and arguments
• Execution duration
• Success/failure status
• Cost attribution

Key Enterprise Benefits

Requirement	AgentHelm Feature
Audit trail	ExecutionTracer + SQLite storage
Distributed tracing	OpenTelemetry + Jaeger
Atomic operations	Saga pattern with compensating tools
Session persistence	MemoryHub with Redis/SQLite
Context search	Semantic memory with Qdrant
Cost control	Built-in pricing for 20+ LLM providers
Human review	Plan YAML export/import workflow

Conclusion

Enterprise AI agent deployments require more than just "an agent that works." They require:

• Traceability for compliance and debugging
• Atomicity for data consistency
• Memory persistence for context continuity

AgentHelm v0.3.0 provides these as first-class features, not afterthoughts.

---

Documentation: hadywalied.github.io/agenthelm

GitHub: github.com/hadywalied/agenthelm