← All Cookbooks
CrewAIIntermediate25 min
CrewAI + HatiData: Chain-of-Thought Audit
Add full chain-of-thought logging to a CrewAI crew for compliance auditing. Replay any agent's decision-making process and export audit reports.
What You'll Build
A CrewAI crew with full chain-of-thought logging for compliance auditing and decision replay.
Prerequisites
$pip install crewai hatidata-agent
$hati init
$OpenAI API key
Architecture
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ CrewAI │───▶│ HatiData │───▶│ Engine │ │ Crew │ │ CoT Ledger │ │ (immutable) │ └──────────────┘ └──────────────┘ └──────────────┘ Agent reasoning → cryptographic hash chain → audit export
Key Concepts
- ●Immutable audit trail: the CoT ledger is append-only with cryptographic hash chaining, so past reasoning cannot be altered
- ●Step categorization: steps are typed (observation, retrieval, analysis, tool_call, conclusion) for structured audit reporting
- ●verify_chain(): cryptographic proof that the audit trail has not been tampered with since creation
- ●Export-ready: SQL aggregations generate audit reports directly from the ledger data
Step-by-Step Implementation
1
Install Dependencies
Install CrewAI and the HatiData agent SDK.
Bash
pip install crewai hatidata-agent
hati initExpected Output
HatiData initialized successfully.
Proxy running on localhost:5439Note: The CoT ledger stores every reasoning step in a tamper-proof, cryptographically hash-chained table.
2
Configure CoT Logging for a Crew
Set up chain-of-thought logging that captures every step of a CrewAI task.
Python
from hatidata_agent import HatiDataAgent
hati = HatiDataAgent(host="localhost", port=5439, agent_id="compliance-crew")
session_id = "compliance-review-001"
# Log the task start
hati.execute(f"""
SELECT log_reasoning_step(
'{session_id}',
'observation',
'Starting compliance review of Q3 vendor contracts'
)
""")
# Log a retrieval step
hati.execute(f"""
SELECT log_reasoning_step(
'{session_id}',
'retrieval',
'Fetched 47 active vendor contracts from contracts table'
)
""")
# Log analysis
hati.execute(f"""
SELECT log_reasoning_step(
'{session_id}',
'analysis',
'Found 3 contracts missing data processing addendums (GDPR requirement)'
)
""")
# Log a tool call
hati.execute(f"""
SELECT log_reasoning_step(
'{session_id}',
'tool_call',
'Queried vendor risk scores: Vendor A (High), Vendor B (Medium), Vendor C (Low)'
)
""")
# Log conclusion
hati.execute(f"""
SELECT log_reasoning_step(
'{session_id}',
'conclusion',
'Recommend immediate DPA amendments for Vendor A (high risk) and Vendor B (medium risk)'
)
""")
print(f"Logged 5 reasoning steps to session {session_id}")Expected Output
Logged 5 reasoning steps to session compliance-review-001Note: Each step is automatically hash-chained to the previous step. The chain cannot be tampered with retroactively.
3
Replay Agent Decisions
Replay the full decision-making process for a compliance audit session.
Python
# Replay the session
trace = hati.query(f"""
SELECT step_number, step_type, content, hash
FROM _hatidata_cot.traces
WHERE session_id = '{session_id}'
ORDER BY step_number ASC
""")
print(f"=== Decision Replay: {session_id} ===")
print(f"Steps: {len(trace)}\n")
for step in trace:
icons = {
"observation": "EYE",
"retrieval": "DB",
"analysis": "CHART",
"tool_call": "TOOL",
"conclusion": "CHECK",
}
icon = icons.get(step['step_type'], "DOT")
print(f" [{icon}] Step {step['step_number']}: {step['step_type']}")
print(f" {step['content']}")
print(f" Hash: {step['hash'][:24]}...")
print()Expected Output
=== Decision Replay: compliance-review-001 ===
Steps: 5
[EYE] Step 1: observation
Starting compliance review of Q3 vendor contracts
Hash: 7a2b3c4d5e6f7a8b9c0d...
[DB] Step 2: retrieval
Fetched 47 active vendor contracts from contracts table
Hash: 8b3c4d5e6f7a8b9c0d1e...
[CHART] Step 3: analysis
Found 3 contracts missing data processing addendums (GDPR requirement)
Hash: 9c4d5e6f7a8b9c0d1e2f...
[TOOL] Step 4: tool_call
Queried vendor risk scores: Vendor A (High), Vendor B (Medium), Vendor C (Low)
Hash: 0d5e6f7a8b9c0d1e2f3a...
[CHECK] Step 5: conclusion
Recommend immediate DPA amendments for Vendor A (high risk) and Vendor B (medium risk)
Hash: 1e6f7a8b9c0d1e2f3a4b...4
Verify Chain and Export Audit Report
Verify the hash chain integrity and export an audit report.
Python
# Verify the chain
verification = hati.query(f"""
SELECT verify_chain('{session_id}') AS is_valid
""")
is_valid = verification[0]['is_valid']
# Build audit report
report = hati.query(f"""
SELECT
session_id,
COUNT(*) AS total_steps,
MIN(created_at) AS started_at,
MAX(created_at) AS completed_at,
COUNT(CASE WHEN step_type = 'conclusion' THEN 1 END) AS conclusions,
COUNT(CASE WHEN step_type = 'tool_call' THEN 1 END) AS tool_calls
FROM _hatidata_cot.traces
WHERE session_id = '{session_id}'
GROUP BY session_id
""")
r = report[0]
print("=== Compliance Audit Report ===")
print(f"Session: {r['session_id']}")
print(f"Hash chain: {'VALID' if is_valid else 'INVALID - TAMPERING DETECTED'}")
print(f"Total steps: {r['total_steps']}")
print(f"Tool calls: {r['tool_calls']}")
print(f"Conclusions: {r['conclusions']}")
print(f"Started: {r['started_at']}")
print(f"Completed: {r['completed_at']}")Expected Output
=== Compliance Audit Report ===
Session: compliance-review-001
Hash chain: VALID
Total steps: 5
Tool calls: 1
Conclusions: 1
Started: 2026-02-28 14:00:00
Completed: 2026-02-28 14:00:03Note: The hash chain verification proves that no steps were inserted, modified, or deleted after logging. Required for regulatory compliance.