Skip to main content

Financial Compliance Assistant Usage Guide

This guide demonstrates how to use the Financial Compliance Assistant (Story 4) for handling regulatory queries with compliance-specific features.

Overview​

The Financial Compliance Assistant is a specialized RAG system designed for financial institutions to help compliance officers navigate complex regulatory requirements. It provides:

  • Regulatory terminology expansion for precise query understanding
  • Authority-weighted reranking to prioritize official guidance
  • Legal disclaimers for all responses
  • Audit logging for compliance tracking
  • Escalation detection for high-risk queries

Quick Start​

Basic Usage​

import asyncio
from packages.agents.compliance_agent import ComplianceAgent

async def basic_compliance_query():
    # Initialize compliance agent
    agent = ComplianceAgent()
    
    # Define user context
    user_context = {
        "user_id": "compliance_officer_001",
        "user_role": "compliance_officer",
        "department": "risk_management",
        "regulation_type": "AML",
        "jurisdiction": "US"
    }
    
    # Query
    query = "What are the reporting requirements for suspicious transactions over $10,000?"
    
    # Handle query
    response = await agent.handle_compliance_query(query, user_context)
    
    print(f"Answer: {response['answer']}")
    print(f"Legal Disclaimers: {response['legal_disclaimers']}")
    print(f"Regulatory Citations: {response['regulatory_citations']}")

# Run
asyncio.run(basic_compliance_query())

Configuration​

Default Configuration​

The compliance assistant comes with sensible defaults optimized for regulatory accuracy:

from packages.rag.compliance_config import ComplianceConfig

config = ComplianceConfig()

print(f"LLM Model: {config.llm_model}")  # gpt-4-turbo-preview
print(f"Temperature: {config.llm_temperature}")  # 0.02 (very low for accuracy)
print(f"Retrieval K: {config.retrieval_k}")  # 30 (more candidates)
print(f"Compliance Level: {config.compliance_level}")  # high

Custom Configuration​

Customize the configuration for specific use cases:

from packages.rag.compliance_config import ComplianceConfig
from packages.agents.compliance_agent import ComplianceAgent

# Create custom config
config = ComplianceConfig(
    llm_temperature=0.01,  # Even lower for maximum accuracy
    retrieval_k=50,  # More retrieval candidates
    rerank_k=15,  # More reranked results
    compliance_level="high",
    enable_audit_logging=True,
    max_cost_per_query=0.30  # Higher cost limit
)

# Create agent with custom config
agent = ComplianceAgent(config)

Use Cases​

1. AML (Anti-Money Laundering) Queries​

async def aml_query_example():
    agent = ComplianceAgent()
    
    user_context = {
        "user_role": "compliance_officer",
        "department": "aml_compliance",
        "regulation_type": "AML",
        "jurisdiction": "US"
    }
    
    query = "What are the SAR filing requirements and deadlines?"
    response = await agent.handle_compliance_query(query, user_context)
    
    # Response includes:
    # - Answer with regulatory details
    # - Legal disclaimers
    # - Regulatory citations (FinCEN, etc.)
    # - Related regulations
    # - Compliance recommendations
    
    return response

2. KYC (Know Your Customer) Queries​

async def kyc_query_example():
    agent = ComplianceAgent()
    
    user_context = {
        "user_role": "compliance_officer",
        "department": "customer_onboarding",
        "regulation_type": "KYC",
        "jurisdiction": "US"
    }
    
    query = "What are the customer due diligence requirements for high-risk customers?"
    response = await agent.handle_compliance_query(query, user_context)
    
    print(f"Answer: {response['answer']}")
    print(f"Related Regulations: {response['related_regulations']}")
    print(f"Recommendations: {response['compliance_recommendations']}")

3. Sanctions Compliance Queries​

async def sanctions_query_example():
    agent = ComplianceAgent()
    
    user_context = {
        "user_role": "compliance_officer",
        "department": "sanctions_screening",
        "regulation_type": "Sanctions",
        "jurisdiction": "US"
    }
    
    query = "What are the OFAC sanctions compliance requirements?"
    response = await agent.handle_compliance_query(query, user_context)
    
    # Check for escalation
    if response.get('escalation_required'):
        print(f"Escalation Info: {response['escalation_info']}")

4. BSA (Bank Secrecy Act) Queries​

async def bsa_query_example():
    agent = ComplianceAgent()
    
    user_context = {
        "user_role": "compliance_officer",
        "department": "transaction_monitoring",
        "regulation_type": "BSA",
        "jurisdiction": "US"
    }
    
    query = "What are the CTR filing requirements?"
    response = await agent.handle_compliance_query(query, user_context)
    
    print(f"Regulatory Context: {response['regulatory_context']}")

5. Tax Compliance (FATCA/CRS) Queries​

async def tax_compliance_query_example():
    agent = ComplianceAgent()
    
    user_context = {
        "user_role": "compliance_officer",
        "department": "tax_compliance",
        "regulation_type": "Tax_Compliance",
        "jurisdiction": "US"
    }
    
    query = "What are the FATCA reporting requirements?"
    response = await agent.handle_compliance_query(query, user_context)

API Endpoints​

Query Endpoint​

import requests

# Compliance query
response = requests.post(
    "http://localhost:8000/compliance/query",
    json={
        "query": "What are the SAR filing requirements?",
        "user_context": {
            "user_role": "compliance_officer",
            "department": "compliance"
        },
        "regulation_type": "AML",
        "jurisdiction": "US",
        "compliance_level": "high"
    },
    headers={"Authorization": "Bearer your-token"}
)

result = response.json()
print(f"Answer: {result['answer']}")
print(f"Query ID: {result['query_id']}")

Search Endpoint​

# Search compliance documents
response = requests.post(
    "http://localhost:8000/compliance/search",
    json={
        "query": "suspicious activity reporting",
        "filters": {
            "regulation_type": "AML",
            "jurisdiction": "US"
        },
        "top_k": 10
    },
    headers={"Authorization": "Bearer your-token"}
)

results = response.json()
print(f"Found {results['total_results']} documents")

Regulatory Lookup Endpoint​

# Look up specific regulation
response = requests.post(
    "http://localhost:8000/compliance/regulatory-lookup",
    json={
        "regulation": "SAR",
        "jurisdiction": "US"
    },
    headers={"Authorization": "Bearer your-token"}
)

regulation_info = response.json()

Response Structure​

Compliance Query Response​

{
    "query_id": "comp_1234567890",
    "answer": "Detailed compliance response...",
    "regulatory_context": {
        "regulation_type": "AML",
        "jurisdiction": "US",
        "compliance_level": "high",
        "last_updated": "2024-10-08T12:00:00Z"
    },
    "legal_disclaimers": [
        "This information is for general guidance only",
        "Specific situations may require legal counsel",
        "Regulations are subject to change",
        "Always consult with compliance legal team for complex cases"
    ],
    "regulatory_citations": ["FinCEN", "SEC"],
    "compliance_metadata": {
        "user_role": "compliance_officer",
        "department": "risk_management",
        "query_timestamp": "2024-10-08T12:00:00Z",
        "compliance_level": "high"
    },
    "compliance_recommendations": [
        "Ensure your AML program is up to date",
        "Review suspicious activity monitoring procedures"
    ],
    "related_regulations": ["BSA", "KYC", "Sanctions"],
    "escalation_required": false,
    "processing_time_ms": 1250.5
}

Escalation Handling​

The compliance assistant automatically detects queries that require human review:

async def escalation_example():
    agent = ComplianceAgent()
    
    # Query that triggers escalation
    query = "I need legal advice on a specific fraud case"
    user_context = {"user_role": "compliance_officer"}
    
    response = await agent.handle_compliance_query(query, user_context)
    
    if response['escalation_required']:
        print(f"Escalation Reason: {response['escalation_info']['reason']}")
        print(f"Priority: {response['escalation_info']['priority']}")
        print(f"Escalation Level: {response['escalation_info']['escalation_level']}")

Escalation Triggers​

Queries are escalated when they contain:

  • Requests for legal advice
  • References to specific cases
  • High-risk terms (fraud, criminal, terrorism)
  • Unclear or conflicting regulations
  • Emergency situations

Audit Logging​

All compliance queries are automatically logged for audit purposes:

# Audit logs include:
{
    "query_id": "comp_1234567890",
    "timestamp": "2024-10-08T12:00:00Z",
    "user_id": "user_123",
    "user_role": "compliance_officer",
    "department": "risk_management",
    "query": "What are the SAR filing requirements?",
    "response_length": 1500,
    "escalation_required": false,
    "processing_time_ms": 1250.5
}

Best Practices​

1. Provide Complete User Context​

user_context = {
    "user_id": "user_123",
    "user_role": "compliance_officer",
    "department": "risk_management",
    "regulation_type": "AML",
    "jurisdiction": "US",
    "compliance_level": "high"
}

2. Use Specific Regulation Types​

  • AML - Anti-Money Laundering
  • KYC - Know Your Customer
  • BSA - Bank Secrecy Act
  • Sanctions - Sanctions Compliance
  • Tax_Compliance - FATCA/CRS
  • General_Compliance - General compliance

Always include legal disclaimers in user-facing applications:

disclaimers = response['legal_disclaimers']
for disclaimer in disclaimers:
    print(f"⚠️ {disclaimer}")

4. Handle Escalations Appropriately​

if response['escalation_required']:
    # Route to human compliance officer
    escalate_to_human(response['escalation_info'])
else:
    # Display response to user
    display_compliance_response(response)

5. Monitor Compliance Metrics​

Track compliance query metrics for continuous improvement:

from packages.observability import MetricsCollector

metrics = MetricsCollector()
metrics.record_compliance_query_metrics(
    query_length=len(query),
    processing_time=processing_time,
    documents_retrieved=len(retrieval_results),
    escalation_required=response['escalation_required']
)

Error Handling​

async def robust_compliance_query():
    agent = ComplianceAgent()
    
    try:
        response = await agent.handle_compliance_query(query, user_context)
        return response
    except Exception as e:
        # Log error
        logger.error(f"Compliance query failed: {e}")
        
        # Return error response with escalation
        return {
            "answer": "I apologize, but I encountered an error...",
            "error": str(e),
            "escalation_required": True
        }

Integration with Existing Systems​

1. Add to FastAPI Application​

from packages.rag.compliance_api import router as compliance_router

app.include_router(compliance_router)

2. Use with Rate Limiting​

from packages.rate_limiting import RateLimitingMiddleware

# Apply rate limiting to compliance endpoints
app.add_middleware(
    RateLimitingMiddleware,
    rate_limit="100/hour",
    burst_limit=10
)

3. Enable Observability​

from packages.observability import LangSmithClient

# Enable tracing for compliance queries
langsmith_client = LangSmithClient(project="compliance-assistant")

Next Steps​

Support​

For questions or issues: