← All Cookbooks
LangChainAdvanced30 min
LangChain + HatiData: Policy Enforcement
Set up HatiData's policy engine to enforce query governance on LangChain agents. Block dangerous queries, log violations, and audit agent behavior.
What You'll Build
A LangChain agent operating under HatiData's policy engine with real-time query governance and audit logging.
Prerequisites
$pip install langchain langchain-openai hatidata-agent
$hati init
$OpenAI API key
Architecture
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ LangChain │───▶│ HatiData │───▶│ Engine │
│ Agent │ │ Policy Eng. │ │ (execute) │
└──────────────┘ └──────┬───────┘ └──────────────┘
│
┌──────▼───────┐
│ Audit Log │
│ (violations) │
└──────────────┘Key Concepts
- ●Policy engine: every query passes through the policy engine before execution, with sub-10ms evaluation latency
- ●Three action types: block (reject), warn (execute + flag), log (execute + audit trail)
- ●Pattern matching: policies use operators like regex, contains, eq, gt for flexible rule definitions
- ●Audit trail: all policy evaluations are logged with timestamp, agent, query, and policy details for compliance reporting
- ●Agent-transparent: agents do not need to know about policies — the proxy enforces them automatically
Step-by-Step Implementation
1
Install Dependencies
Install LangChain and the HatiData agent SDK.
Bash
pip install langchain langchain-openai hatidata-agent
hati initExpected Output
HatiData initialized successfully.
Proxy running on localhost:5439Note: The policy engine runs inside the HatiData proxy and evaluates every query before execution.
2
Define Policies
Create policies that govern what agents can and cannot do.
Python
from hatidata_agent import HatiDataAgent
hati = HatiDataAgent(host="localhost", port=5439, agent_id="policy-admin")
# Policy 1: Block DROP and TRUNCATE statements
hati.execute("""
INSERT INTO _hatidata_policies.rules VALUES (
'no-destructive-ops',
'block',
'statement_type',
'regex',
'^(DROP|TRUNCATE)',
'Destructive operations are not allowed'
)
""")
# Policy 2: Warn on queries that scan more than 1M rows
hati.execute("""
INSERT INTO _hatidata_policies.rules VALUES (
'large-scan-warning',
'warn',
'estimated_rows',
'gt',
'1000000',
'Query scans more than 1M rows — consider adding filters'
)
""")
# Policy 3: Log all queries touching PII tables
hati.execute("""
INSERT INTO _hatidata_policies.rules VALUES (
'pii-access-audit',
'log',
'tables_accessed',
'contains',
'customers,employees',
'Query accesses PII table'
)
""")
print("Policies configured:")
print(" 1. no-destructive-ops: BLOCK DROP/TRUNCATE")
print(" 2. large-scan-warning: WARN on >1M row scans")
print(" 3. pii-access-audit: LOG PII table access")Expected Output
Policies configured:
1. no-destructive-ops: BLOCK DROP/TRUNCATE
2. large-scan-warning: WARN on >1M row scans
3. pii-access-audit: LOG PII table accessNote: Policy actions: block (reject query), warn (execute but flag), log (execute and record for audit).
3
Configure the LangChain Agent with Proxy
Point your LangChain agent at HatiData's proxy so all queries go through the policy engine.
Python
from langchain_openai import ChatOpenAI
from hatidata_agent import HatiDataAgent
llm = ChatOpenAI(model="gpt-4o")
hati = HatiDataAgent(host="localhost", port=5439, agent_id="governed-agent")
def safe_query(sql: str) -> dict:
"""Execute a query through the policy engine."""
try:
result = hati.query(sql)
return {"status": "ok", "data": result}
except Exception as e:
return {"status": "blocked", "error": str(e)}
# Test a safe query
result = safe_query("SELECT COUNT(*) AS cnt FROM products")
print(f"Safe query: {result['status']}")
print(f" Result: {result['data']}")Expected Output
Safe query: ok
Result: [{'cnt': 42}]4
Test Blocked Queries
Verify that destructive queries are blocked by the policy engine.
Python
# Test a destructive query — should be blocked
result = safe_query("DROP TABLE customers")
print(f"DROP TABLE: {result['status']}")
print(f" Error: {result['error']}")
# Test another blocked query
result = safe_query("TRUNCATE TABLE employees")
print(f"\nTRUNCATE: {result['status']}")
print(f" Error: {result['error']}")
# Test a safe query that accesses PII table (logged, not blocked)
result = safe_query("SELECT name, email FROM customers LIMIT 5")
print(f"\nPII query: {result['status']}")
print(f" Result: {len(result['data'])} rows (logged for audit)")Expected Output
DROP TABLE: blocked
Error: Policy violation: Destructive operations are not allowed [no-destructive-ops]
TRUNCATE: blocked
Error: Policy violation: Destructive operations are not allowed [no-destructive-ops]
PII query: ok
Result: 5 rows (logged for audit)Note: Blocked queries return an error immediately. Logged queries execute normally but create an audit trail.
5
Audit Policy Violations
Query the audit log to review all policy evaluations and violations.
Python
# Query the audit log
audit_log = hati.query("""
SELECT timestamp, agent_id, policy_name, action,
query_preview, message
FROM _hatidata_audit.policy_log
ORDER BY timestamp DESC
LIMIT 10
""")
print("=== Policy Audit Log ===")
for entry in audit_log:
icon = {"block": "BLOCKED", "warn": "WARNING", "log": "LOGGED"}[entry['action']]
print(f" [{icon}] {entry['timestamp']}")
print(f" Agent: {entry['agent_id']}")
print(f" Policy: {entry['policy_name']}")
print(f" Query: {entry['query_preview'][:60]}")
print(f" Message: {entry['message']}")
print()Expected Output
=== Policy Audit Log ===
[LOGGED] 2026-02-28 14:32:15
Agent: governed-agent
Policy: pii-access-audit
Query: SELECT name, email FROM customers LIMIT 5
Message: Query accesses PII table
[BLOCKED] 2026-02-28 14:32:10
Agent: governed-agent
Policy: no-destructive-ops
Query: TRUNCATE TABLE employees
Message: Destructive operations are not allowed
[BLOCKED] 2026-02-28 14:32:05
Agent: governed-agent
Policy: no-destructive-ops
Query: DROP TABLE customers
Message: Destructive operations are not allowed