CrewAI + HatiData: Credit-Aware Agents

pip install crewai hatidata-agent
hati init

HatiData initialized successfully.
Proxy running on localhost:5439

from hatidata_agent import HatiDataAgent

hati = HatiDataAgent(host="localhost", port=5439, agent_id="budget-manager")

# Check current usage
usage = hati.query("""
    SELECT
        org_id,
        monthly_credit_limit,
        used_credits,
        monthly_credit_limit - used_credits AS remaining,
        (used_credits::DOUBLE / monthly_credit_limit * 100) AS usage_pct
    FROM _hatidata_metering.quotas
    WHERE org_id = 'default'
""")

u = usage[0]
print("=== Credit Usage ===")
print(f"  Monthly limit: {u['monthly_credit_limit']} credits")
print(f"  Used: {u['used_credits']} credits")
print(f"  Remaining: {u['remaining']} credits")
print(f"  Usage: {u['usage_pct']:.1f}%")

=== Credit Usage ===
  Monthly limit: 10000 credits
  Used: 3247 credits
  Remaining: 6753 credits
  Usage: 32.5%

# Get per-agent usage breakdown
breakdown = hati.query("""
    SELECT
        agent_id,
        COUNT(*) AS query_count,
        SUM(credits_used) AS total_credits,
        AVG(credits_used) AS avg_credits_per_query,
        MAX(credits_used) AS max_single_query
    FROM _hatidata_metering.usage_log
    WHERE timestamp > CURRENT_DATE - INTERVAL '7 days'
    GROUP BY agent_id
    ORDER BY total_credits DESC
""")

print("=== Per-Agent Credit Usage (7 days) ===")
for b in breakdown:
    print(f"  {b['agent_id']}:")
    print(f"    Queries: {b['query_count']}, Total credits: {b['total_credits']}")
    print(f"    Avg/query: {b['avg_credits_per_query']:.1f}, Max single: {b['max_single_query']}")

# Get cost by query type
cost_by_type = hati.query("""
    SELECT
        query_type,
        COUNT(*) AS count,
        SUM(credits_used) AS total_credits
    FROM _hatidata_metering.usage_log
    WHERE timestamp > CURRENT_DATE - INTERVAL '7 days'
    GROUP BY query_type
    ORDER BY total_credits DESC
""")

print("\n=== Cost by Query Type ===")
for c in cost_by_type:
    print(f"  {c['query_type']}: {c['count']} queries, {c['total_credits']} credits")

=== Per-Agent Credit Usage (7 days) ===
  research-agent:
    Queries: 156, Total credits: 1890
    Avg/query: 12.1, Max single: 45
  analysis-agent:
    Queries: 89, Total credits: 780
    Avg/query: 8.8, Max single: 32
  writer-agent:
    Queries: 43, Total credits: 577
    Avg/query: 13.4, Max single: 28

=== Cost by Query Type ===
  vector_search: 145 queries, 1740 credits
  sql_query: 112 queries, 1120 credits
  store_memory: 31 queries, 387 credits

def get_remaining_budget() -> dict:
    result = hati.query("""
        SELECT
            monthly_credit_limit - used_credits AS remaining,
            (used_credits::DOUBLE / monthly_credit_limit * 100) AS usage_pct
        FROM _hatidata_metering.quotas
        WHERE org_id = 'default'
    """)
    return result[0]

def cost_aware_query(sql: str, agent_id: str) -> list:
    budget = get_remaining_budget()

    if budget['usage_pct'] > 90:
        print(f"  [{agent_id}] BUDGET CRITICAL ({budget['usage_pct']:.0f}%) — skipping non-essential queries")
        return []

    if budget['usage_pct'] > 75:
        # Add LIMIT to expensive queries to reduce cost
        if "LIMIT" not in sql.upper():
            sql += " LIMIT 100"
        print(f"  [{agent_id}] BUDGET WARNING ({budget['usage_pct']:.0f}%) — limiting results")

    return hati.query(sql)

# Test cost-aware behavior
print("=== Cost-Aware Agent ===")
result = cost_aware_query(
    "SELECT * FROM _hatidata_memory.memories WHERE namespace = 'research'",
    "research-agent"
)
print(f"  Returned {len(result)} results")

# Store the budget check as a memory for the agent
hati.execute(f"""
    SELECT store_memory(
        'Budget check: {get_remaining_budget()["remaining"]} credits remaining. Adapting query strategy.',
        'cost-tracking'
    )
""")

=== Cost-Aware Agent ===
  [research-agent] BUDGET WARNING (78%) — limiting results
  Returned 100 results

# Register a trigger for budget discussions
hati.execute("""
    SELECT register_trigger(
        'budget-alert',
        'credit usage approaching limit, budget running low, cost overrun',
        0.7,
        'webhook',
        'https://your-app.com/hooks/budget-alert'
    )
""")

print("Budget alert trigger registered")

# Check if we should alert right now
budget = get_remaining_budget()
alert_thresholds = [
    (90, "CRITICAL"),
    (75, "WARNING"),
    (50, "INFO"),
]

for threshold, level in alert_thresholds:
    if budget['usage_pct'] >= threshold:
        print(f"\n[{level}] Credit usage at {budget['usage_pct']:.1f}%")
        print(f"  Remaining: {budget['remaining']} credits")
        if level == "CRITICAL":
            print("  Action: Switching all agents to minimal mode")
        elif level == "WARNING":
            print("  Action: Limiting query result sizes")
        break
else:
    print(f"\n[OK] Credit usage at {budget['usage_pct']:.1f}% — within budget")

Budget alert trigger registered

[WARNING] Credit usage at 78.2%
  Remaining: 2180 credits
  Action: Limiting query result sizes