LangChainIntermediate25 min

LangChain + HatiData: CoT Replay & Audit

Log chain-of-thought reasoning from LangChain agents into HatiData's tamper-proof ledger. Replay sessions and verify hash chain integrity for compliance auditing.

What You'll Build

A LangChain agent that logs chain-of-thought reasoning to HatiData's tamper-proof ledger, with session replay and hash chain verification.

Prerequisites

$pip install langchain langchain-openai hatidata-agent

$hati init

$OpenAI API key

Architecture

┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│  LangChain   │───▶│  HatiData    │───▶│   Engine     │
│   Agent      │    │  CoT Ledger  │    │ (immutable)  │
└──────────────┘    └──────────────┘    └──────────────┘
   Reasoning steps → cryptographic hash-chained ledger

Key Concepts

●Immutable hash chain: each reasoning step's cryptographic hash includes the previous step's hash, creating a tamper-proof audit trail
●Session replay: replay any session's complete chain-of-thought to understand exactly how the agent reached its conclusions
●Step types: categorize reasoning steps (observation, retrieval, analysis, reasoning, conclusion) for structured auditing
●Compliance-ready: the append-only ledger satisfies regulatory requirements for AI decision audit trails
●verify_chain(): cryptographically verify that no steps have been inserted, modified, or deleted after the fact

Step-by-Step Implementation

Install Dependencies

Install LangChain and the HatiData agent SDK with CoT support.

Bash

pip install langchain langchain-openai hatidata-agent
hati init

Expected Output

HatiData initialized successfully.
Proxy running on localhost:5439
MCP server running on localhost:5440

Note: hati init starts the proxy with the CoT ledger enabled by default. All traces are stored in _hatidata_cot tables.

Configure CoT Logging

Set up the HatiData agent to log reasoning steps from your LangChain chain.

Python

from hatidata_agent import HatiDataAgent

hati = HatiDataAgent(host="localhost", port=5439, agent_id="langchain-auditor")

# Start a new CoT session
session_id = "audit-session-001"

# Log the first reasoning step
hati.execute(f"""
    SELECT log_reasoning_step(
        '{session_id}',
        'observation',
        'User asked about Q3 revenue trends for APAC region'
    )
""")

print(f"CoT session started: {session_id}")

Expected Output

CoT session started: audit-session-001

Note: Each session gets its own cryptographic hash chain. Steps are append-only and tamper-proof.

Log Multi-Step Reasoning

Log a complete reasoning chain as your LangChain agent processes a query.

Python

# Log sequential reasoning steps
steps = [
    ("retrieval", "Fetching Q3 APAC revenue data from sales table"),
    ("analysis", "APAC revenue grew 34% YoY, driven by Indonesia (+52%) and Vietnam (+41%)"),
    ("reasoning", "Growth correlates with cloud infrastructure expansion in SEA markets"),
    ("conclusion", "Recommend increasing APAC sales team headcount by 40% for Q4"),
]

for step_type, content in steps:
    hati.execute(f"""
        SELECT log_reasoning_step(
            '{session_id}',
            '{step_type}',
            '{content}'
        )
    """)
    print(f"  Logged: [{step_type}] {content[:60]}...")

print(f"\nLogged {len(steps)} reasoning steps to session {session_id}")

Expected Output

  Logged: [retrieval] Fetching Q3 APAC revenue data from sales table...
  Logged: [analysis] APAC revenue grew 34% YoY, driven by Indonesia (+52...
  Logged: [reasoning] Growth correlates with cloud infrastructure expansi...
  Logged: [conclusion] Recommend increasing APAC sales team headcount by ...

Logged 4 reasoning steps to session audit-session-001

Note: Step types include: observation, retrieval, analysis, reasoning, conclusion, tool_call, error, and more.

Replay a Session

Replay the full chain-of-thought for a session to audit the agent's decision-making process.

Python

# Replay the entire session
trace = hati.query(f"""
    SELECT step_number, step_type, content, hash, created_at
    FROM _hatidata_cot.traces
    WHERE session_id = '{session_id}'
    ORDER BY step_number ASC
""")

print(f"=== CoT Replay: {session_id} ===")
print(f"Total steps: {len(trace)}\n")
for step in trace:
    print(f"Step {step['step_number']}: [{step['step_type']}]")
    print(f"  {step['content']}")
    print(f"  Hash: {step['hash'][:16]}...")
    print()

Expected Output

=== CoT Replay: audit-session-001 ===
Total steps: 5

Step 1: [observation]
  User asked about Q3 revenue trends for APAC region
  Hash: a3f8b2c1d4e5f6a7...

Step 2: [retrieval]
  Fetching Q3 APAC revenue data from sales table
  Hash: b7c9d0e1f2a3b4c5...

Step 3: [analysis]
  APAC revenue grew 34% YoY, driven by Indonesia (+52%) and Vietnam (+41%)
  Hash: c1d2e3f4a5b6c7d8...

Verify Chain Integrity

Verify the cryptographic hash chain to ensure no reasoning steps have been tampered with.

Python

# Verify the hash chain integrity
verification = hati.query(f"""
    SELECT verify_chain('{session_id}') AS is_valid
""")

is_valid = verification[0]['is_valid']
print(f"Hash chain verification: {'PASSED' if is_valid else 'FAILED'}")
print(f"Session: {session_id}")
print(f"Chain type: cryptographic append-only")
print(f"Tamper detection: {'No tampering detected' if is_valid else 'WARNING: Chain broken!'}")

Expected Output

Hash chain verification: PASSED
Session: audit-session-001
Chain type: cryptographic append-only
Tamper detection: No tampering detected

Note: Each step's hash includes the previous step's hash, forming an unbreakable chain. Any modification to a past step invalidates all subsequent hashes.

Related Use Case

Compliance

Autonomous Compliance Auditor

Continuous Compliance, Not Annual Audits

Ready to build?

Install HatiData locally and start building with LangChain in minutes.

Join Waitlist