AI in CareerForge
Deep Dive

What AI Does in CareerForge

AI in CareerForge is not a single feature — it is an intelligence layer that runs across all four portals, generating personalised career guidance, synthesising market data, and building structured learning paths. Every AI call is grounded in real business context (the employee's profile, skill inventory, anonymised coaching relationship) and constrained to JSON-structured output.

AI Architecture

The AI layer is a dedicated microservice (ai-connector-service, port 8001) plus two Celery workers (celery-worker, celery-beat). All other backend services import directly from the shared RAG module — there is no inter-service HTTP call to the AI connector for inference; only for connector management (CRUD) and async task status polling.

Component Map

LLM Provider — AI Workflow Connector

The LLM provider is no longer hardcoded. Admins configure an AI Workflow Config in the Admin portal (Connectors → AI Workflow tab). The active config is loaded from PostgreSQL with a 5-minute in-process cache. Supported providers:

The _ai_chat() dispatcher reads the active workflow config and routes to the correct provider. For self-hosted, it calls an OpenAI-compatible /v1/chat/completions endpoint at the configured base_url (auto-appends /v1 if missing; 120s timeout):

async def _ai_chat(prompt: str, system: str = "") -> str:
    # rag.py — routes to active AI Workflow Config provider
    workflow = await _get_active_workflow()
    provider = workflow.get("provider", "anthropic")
    if provider == "anthropic":   return await _anthropic_chat(...)
    elif provider == "openai":    return await _openai_chat(...)
    elif provider == "perplexity": return await _perplexity_chat(...)
    elif provider == "gemini":    return await _gemini_chat(...)
    elif provider == "self_hosted":return await _self_hosted_chat(..., base_url=workflow["base_url"])
    return ""

Component Map

AI Workflow Config is also stored in PostgreSQL (ai_workflow_configs table) alongside the existing ai_documents and connector tables. The ai_suggestion_logs table records every AI call with portal, feature, pseudo-id, model used, and expiry.

Embedding Provider Fallback

Both providers produce 1536-dimensional vectors, stored in PostgreSQL with the pgvector extension.

The RAG Pattern

RAG — Retrieval-Augmented Generation — grounds LLM responses in real documents. In CareerForge the knowledge base is the ai_documents table, populated by connector ingestion. Every AI function first searches this table, prepends relevant chunks to the prompt, then calls the LLM.

Request Flow

Public AI Functions

Caching Strategy

AI responses are expensive — both in latency (1–5 seconds) and API cost. CareerForge uses the ai_documents PostgreSQL table as a key-value cache. This gives the cache an automatic audit trail: every cache entry is a document with an ingestion timestamp.

TTL Tiers (by volatility) — now admin-configurable per feature

TTLs for AI suggestion results (audit log expiry + cache) are configured per-feature in the active AI Workflow Config. The defaults below reflect what the system ships with, but any admin can override them without a code deploy:

Cache Key Design

# rag.py:120 — cache lookup with TTL check
async def _get_cached_response(cache_key: str, ttl_hours: int) -> list | dict | None:
    async with AsyncSessionLocal() as session:
        row = (await session.execute(
            sql_text("SELECT content, ingested_at FROM ai_documents "
                     "WHERE source_name = :key ORDER BY ingested_at DESC LIMIT 1"),
            {"key": cache_key}
        )).fetchone()

    if not row:
        return None
    gen_at = datetime.fromisoformat(row.ingested_at)
    if (datetime.now(UTC) - gen_at) > timedelta(hours=ttl_hours):
        return None              # expired — caller will regenerate
    return json.loads(row.content)

AI-Ops & Cost Resilience

A second cost & reliability layer sits in front of every conversational AI endpoint. The goal: shrink the input token bill on every provider, swallow duplicate requests for free, and stop a tripped provider from bleeding the whole replica fleet. Provider-specific optimisations (Anthropic prompt caching) sit behind a capability predicate so deployments on Perplexity / OpenAI / Ollama get the universal wins without any opt-in.

Five layers, in order

Cache key construction

# services/ai_connector/cache.py
def make_cache_key(
    *, pseudo_id: str, question: str,
    workflow_id: str | None, skill_count: int,
    opp_top_ids: list[str], extra: list[str] | None = None,
) -> str:
    parts = [pseudo_id, question.strip().lower(), workflow_id or "",
             str(skill_count), ",".join(sorted(opp_top_ids))]
    if extra: parts.extend(extra)
    digest = hashlib.sha256("|".join(parts).encode()).hexdigest()
    return f"cf:airesp:{digest}"

/ai-chat additionally folds a hash of the recent conversation turns into extra= so a follow-up question with the same words doesn't collide with the original.

Provider-portability principle

Layers 1–4 ship benefits regardless of which AI provider the deployment uses. Layer 5 sits behind a capability predicate that returns False on Perplexity, OpenAI (which auto-caches server-side), Ollama, and Gemini. The same call site path will extend to Gemini's CachedContent by adding one entry to the predicate and a body-rewrite branch in _gemini_chat — no caller changes needed.

Daily quota visible to the user

Phase 1's daily-quota counter is enforced server-side, but historically the user couldn't see it — the first sign that you'd hit the limit was a 429 on your next AI call. GET /api/v1/quota/me exposes the same Redis counter the breaker enforces (no parallel state, no drift) and the employee portal sidebar polls it every 60s. The chip turns orange at ≤ 5 remaining so you get a heads-up before the ceiling.

# services/employee/routers/quota.py
@router.get("/quota/me")
async def get_my_quota(...) -> dict[str, Any]:
    pseudo_id = await _resolve_from_keycloak_sub(...)
    remaining = await remaining_daily_quota(pseudo_id)
    return {
        "remaining": remaining,        # -1 if Redis is unreachable
        "limit":     DEFAULT_DAILY_LIMIT, # 50 / day default
        "resets_at": _next_utc_midnight(),
    }

Same fail-open posture as the rest of Phase 1: when Redis is down the chip just disappears (remaining = -1) rather than lying about state.

Drift gate (two-gate setup)

CI runs both gates in sequence, paired so neither bug class can slip through:

• alembic check — fails when a SQLModel field has no matching migration column. The bug class that manifests as UndefinedColumn 500s after a fresh deploy. migrations/env.py uses an _include_object filter that skips index, FK, and unique-constraint naming divergence (cosmetic only) so this gate flags real drift cleanly.
• pytest test_raw_sql_ghost_tables.py — fails when a router uses raw SQL against a table that has no SQLModel at all. The bug class behind the api_keys incident (table referenced via FROM api_keys, no model, no migration → 500 on first admin click). The alembic gate can't see those tables; this one tokenises every .py file in services/ and asserts each raw-SQL identifier maps to a real table or a tiny allowlist.

Connector Ingestion

Connectors are configurable data sources — REST APIs, RSS feeds, JSON endpoints — that feed real-world market intelligence into CareerForge. The Admin portal provides CRUD for connectors. Ingestion runs hourly via Celery Beat.

Data Models

Ingestion Pipeline (tasks/ingestion.py)

Response normalisation handles multiple API schemas automatically: items[], data[], results[], or bare dict. Field aliases resolve content/body/description/text and url/link transparently.

Celery Pipeline Tasks

Long-running AI generation — learning paths and the weekly digest — runs asynchronously in Celery workers, so API responses remain fast. The client polls GET /api/v1/tasks/{task_id} for status.

Learning Path Output Schema

{
  "target_role": "Senior Data Engineer",
  "summary": "12-month path targeting 8 skills across 4 phases...",
  "nodes": [
    {
      "id": "n1",
      "title": "Apache Kafka Fundamentals",
      "type": "skill",            // skill | cert | project | milestone
      "description": "...",
      "estimated_hours": 40,
      "prerequisites": [],
      "resources": [{ "type": "course", "title": "...", "url": "..." }]
    }
  ]
}

Intelligence Digest Schema

{
  "week_summary": "...",
  "top_skills_rising": [{ "skill": "Rust", "trend": "growing 34% YoY" }],
  "role_movements": [{ "role": "AI Engineer", "direction": "rising", "insight": "..." }],
  "recommended_actions": ["Upskill Python ML team in LangGraph", "..."],
  "generated_at": "2026-04-21T08:00:00Z"
}

Code Walkthrough

The following are the key implementation files with their roles and the most important code patterns.

Prompt Engineering Pattern

All prompts follow a consistent 3-part template: identity declaration → context injection → JSON-only output constraint.

# rag.py:253–265 — role suggestions prompt assembly
system = (
    "You are a career intelligence system specialising in IT career paths. "
    "Suggest emerging roles based on current 2025 market data. "
    "Respond ONLY with a valid JSON array of objects with keys: "
    "role_title (string), match_score (0.0-1.0), description (string), "
    "required_skills (array of strings)."
)
context_block = ("Connector market data:\n" + context + "\n\n") if context else ""
prompt = (
    context_block
    + f"Suggest 5 emerging IT career roles for a {seniority} {specialization} professional "
    + "in 2025. Focus on high-demand, growing roles. Return JSON array only."
)

JSON Parsing with Graceful Fallback

# rag.py:81 — robust JSON extraction from LLM output
def _parse_json(text: str) -> list:
    if not text: return []
    text = text.strip()
    if "```" in text:                        # strip markdown code fences
        for part in text.split("```"):
            p = part.strip()
            if p.startswith("json"): p = p[4:].strip()
            if p.startswith("[") or p.startswith("{"): text = p; break
    try:
        result = json.loads(text)
        return result if isinstance(result, list) else [result]
    except json.JSONDecodeError:
        m = re.search(r"\[.*\]", text, re.DOTALL)  # last-chance regex
        if m:
            try: return json.loads(m.group())
            except: pass
    return []                                      # graceful empty on failure

Safety & Guardrails

Guardrails in CareerForge operate at three levels: prompt-level constraints, input validation, and infrastructure-level access control. The goal is to prevent PII leaks, hallucinated identities, runaway API spend, and prompt injection from malicious connector data.

Role-Based AI Access

Keycloak enforces which user roles can access which AI features. Each portal authenticates against its own realm; the token's realm_access.roles claim is verified on every request.

PII Protection

Personal data is protected at two levels: pseudonymisation of the employee identity before it reaches any AI prompt, and AES-256-GCM encryption of the stored PII fields.

Pseudonymisation Flow

Coaches see only the pseudo_id, the employee's specialisation, seniority, experience years, and region (not city). The HMAC key is injected from Sealed Secrets — not derivable from any data in the LLM prompt.

At-rest Encryption

Employee PII fields (name, email, exact location) are stored encrypted in employee.encrypted_pii using AES-256-GCM. The key is injected at runtime from DEV_PII_KEY_HEX (Vault-backed in production, Sealed Secret in the demo cluster).

Strengths & Weaknesses

An honest assessment of what the current AI implementation does well and where it falls short.

Company Simulation

CareerForge is seeded with 14 realistic personas across 8 engineering practices, spanning 5 seniority levels and 4 user roles. This creates a miniature simulation of a real outsourcing company's internal dynamics — skill distributions, coaching relationships, cohort readiness, and management visibility.

Simulated Population

Dynamics the Simulation Captures

• Skill distribution heterogeneity — a junior QA engineer has a different AI-generated learning path than a senior Data Engineer, even within the same company
• Coach caseload variation — some coaches cover two practices; their AI advice calls span different specialisations, stress-testing the system prompt's generalism
• Temporal dynamics — caches expire at different rates; the snapshot of "current" AI output shifts over the 2h/6h/24h/7d cycle even with no user action
• Commitment and sign-off workflow — forms create durable commitments that accumulate; over time the DB grows a history of what was promised vs delivered
• Manager visibility lag — the manager sees readiness scores that reflect cached AI output; there is a natural delay between a skill being updated and the manager's dashboard updating
• Practice-level cohort patterns — employees in the same practice tend to receive similar role suggestions (same specialisation cache key), surfacing whether the LLM's recommendations are varied enough

Reconciliation Loops

A reconciliation loop is a closed feedback cycle: the platform observes the outcome of its AI recommendations, compares them to what actually happened, and adjusts future recommendations accordingly. CareerForge has the infrastructure to support several such loops — some partially built, some requiring additional work.

The Current Data Signals

Loop 1 — Connector Quality Feedback

Current state: fully implemented infrastructure, loop not closed.

Every connector source has a trend_score field and last_status_code. The ingestion task records IngestionHistory.chunks_ingested and success. A reconciliation job could:

1. Correlate connector sources with AI output quality (do documents from source X lead to higher-confidence market trends?)
2. Automatically lower trend_score for sources that repeatedly 404 or return empty content
3. Promote sources whose topics appear frequently in accepted learning paths

Loop 2 — Learning Path Acceptance Rate

Current state: data exists, loop not implemented.

The learning_path_nodes table records which nodes employees mark complete. A weekly digest job could compute:

-- Proposed reconciliation query
SELECT node_type, skill_name,
       COUNT(*) FILTER (WHERE status = 'completed') AS completed,
       COUNT(*) FILTER (WHERE status = 'skipped')  AS skipped,
       COUNT(*) AS total
FROM learning_path_nodes
WHERE created_at > NOW() - INTERVAL '30 days'
GROUP BY node_type, skill_name
ORDER BY (skipped::float / NULLIF(total, 0)) DESC;

Nodes with high skip rates could feed back into the system prompt: "Avoid recommending X for specialisation at seniority level — historically low completion rate."

Loop 3 — Role Suggestion Relevance

Current state: data partially exists.

If an employee clicks through a role suggestion and sets it as their target role, that is an acceptance signal. The ratio of suggested-to-accepted roles, broken down by specialisation and seniority, tells us whether the LLM is recommending roles people actually want. This can be used to:

• Adjust the role suggestion prompt to reflect which roles in which practices actually get accepted
• Build a fine-tuning dataset of (profile → accepted role) pairs
• Invalidate the 24h cache early for a sub-segment if the acceptance rate drops below a threshold

Loop 4 — Coaching Advice Quality Score

Current state: not implemented. Requires UI addition.

Add a thumbs-up/down widget after each AI coaching response. Store the score in a new ai_feedback table keyed by (pseudo_id, question_hash, model_used, score). The reconciliation job then:

1. Computes per-model quality scores over rolling 30-day windows
2. Routes to the higher-scoring model when both are available
3. Feeds low-scored (question, answer) pairs back to the prompt as negative examples

Loop 5 — Market Trend Accuracy

Current state: infrastructure exists for trend_direction tracking. Loop not closed.

Perplexity sonar-pro returns real-time web search context. Store the predicted trend_direction at time T. At time T+30d, re-run the trend query and compare. If the direction is consistent, increase the confidence score for that topic. If it reversed, lower it and flag for human review. Over time this builds a calibration curve for the AI's trend forecasting ability.

Self-Improvement Architecture (Target State)

# Proposed: weekly reconciliation task
@celery_app.task(name="reconcile_ai_quality")
async def reconcile_ai_quality():
    signals = {
        "path_skip_rates":   await compute_node_skip_rates(),
        "role_accept_rates": await compute_role_acceptance(),
        "advice_scores":     await aggregate_feedback_scores(),
        "connector_quality": await score_connector_sources(),
    }
    # Write signals as a new ai_document for RAG context injection
    await store_cached_response("__reconciliation_signals__", signals)
    # Update connector trend_scores based on quality signals
    await update_connector_scores(signals["connector_quality"])
    # Invalidate stale caches for sub-segments with poor outcomes
    await invalidate_poor_performing_caches(signals)

Future Roadmap

The following developments are ordered by impact and implementation complexity. Each builds on the existing infrastructure without requiring architectural rewrites.