BLOCKLOG

End-to-End Example

AI Agent Incident Reconstruction

Walkthrough of a multi-agent hedge fund workflow where a risk limit violation triggers a human-in-the-loop review and downstream forensic replay.

Workflow

Analyst → Risk → Executor

Trigger

Risk limit violation

Outcome

HITL review + forensic replay

Step 1

Multi-agent setup

An analyst agent produces a trading signal, a risk agent checks portfolio limits, and an executor fills the trade. A single WORKFLOW_ID binds the execution context across the workflow so every decision and tool call can be reconstructed later.

import os
import random
from uuid import uuid4
import blocklog

# Initialize SDK
blocklog.init(api_key=os.environ.get("BLOCKLOG_API_KEY", "blk_demo_key"))
WORKFLOW_ID = str(uuid4())

@blocklog.tool(name="fetch-price")
def fetch_price(ticker: str) -> float:
    return {"TSLA": 412.50, "AAPL": 189.30}.get(ticker, 250.0)

@blocklog.tool(name="check-risk-limits")
def check_risk_limits(ticker: str, qty: int, price: float) -> dict:
    trade_value = qty * price
    return {
        "approved": trade_value < 50_000,
        "trade_value": trade_value,
        "limit": 50_000,
    }

@blocklog.agent(name="market-analyst", version="2.1")
def analyst_agent(ticker: str) -> dict:
    price = fetch_price(ticker)
    score = 0.85
    signal = "BUY"

    with blocklog.decision(
        type="SIGNAL",
        asset=ticker,
        confidence=score,
        metadata={"workflow_id": WORKFLOW_ID},
    ) as d:
        d.record_input(price=price)
        d.record_output(signal=signal, score=score)

    return {"ticker": ticker, "price": price, "signal": signal, "decision_id": d.id}

@blocklog.agent(name="risk-manager", version="1.5")
def risk_agent(ticker: str, price: float, qty: int, analyst_dec_id: str) -> dict:
    risk = check_risk_limits(ticker, qty, price)

    with blocklog.decision(
        type="RISK_APPROVAL",
        asset=ticker,
        confidence=1.0 if risk["approved"] else 0.0,
        metadata={"workflow_id": WORKFLOW_ID},
    ) as d:
        d.record_input(qty=qty, price=price, analyst_decision_id=analyst_dec_id)
        d.record_output(approved=risk["approved"])

        if not risk["approved"]:
            d.request_approval(
                reason=f"Value ${risk['trade_value']:.0f} exceeds limit ${risk['limit']:.0f}",
                reviewer="cro@fund.com",
            )

    return {**risk, "risk_decision_id": d.id}

if __name__ == "__main__":
    TICKER = "TSLA"
    QTY = 150  # 150 * 412.50 = $61,875

    analysis = analyst_agent(TICKER)
    risk_status = risk_agent(TICKER, analysis["price"], QTY, analysis["decision_id"])

Step 2

Query forensics and root cause

Once an exception or human review request is raised, the forensic replay API reconstructs the timeline, surfaces the causal chain, and helps identify the root cause behind the incident.

import blocklog

session = blocklog.replay(trace_id="your-trace-id-here")

for event in session.timeline():
    print(f"[{event.get('at')}] {event.get('item_type')}: {event.get('summary')}")

cause = session.root_cause()
if cause["detected"]:
    print(f"Incident: {cause['root_cause_type']}")
    print(f"Explanation: {cause['description']}")
    print(f"Remediation: {cause['remediation']}")

What this example shows

  • A shared workflow identifier links the signal, risk evaluation, and review path into one auditable trail.
  • Risk approval becomes an explicit decision object instead of an implicit internal check.
  • Human escalation and replay investigation are part of the same forensic lifecycle.