The Connective Tissue of an AI Platform: Workflow, Taxonomy, Auth, and Memory

查看原文 / View Original 编辑 / Edit

The Connective Tissue of an AI Platform: Workflow, Taxonomy, Auth, and Memory

AI 平台的连接组织：工作流、分类法、身份验证与记忆

When you’re building an AI evaluation platform with multiple microservices, the “core” services get all the attention — the evaluation engine, the scoring system, the RAG pipeline. But a platform doesn’t work without the connective tissue: the workflow orchestration that keeps humans in the loop, the taxonomy engine that classifies tasks intelligently, the platform service that ties authentication together, and the evaluation suites that ensure models actually remember context. These four services don’t make headlines, but they’re what turned a collection of microservices into an actual platform. Here’s what went into each one and why the engineering decisions mattered.

当你构建一个包含多个微服务的 AI 评估平台时，人们往往只关注“核心”服务——如评估引擎、评分系统和 RAG 流水线。但如果没有连接组织，平台就无法运作：这些组织包括保持“人在回路”（human-in-the-loop）的工作流编排、智能分类任务的分类法引擎、整合身份验证的平台服务，以及确保模型确实能记住上下文的评估套件。这四项服务虽然不会登上头条，但正是它们将一堆微服务变成了一个真正的平台。以下是每一项服务的构建内容以及这些工程决策背后的考量。

Workflow Orchestration: The Human-in-the-Loop Engine

工作流编排：人在回路的引擎

AI evaluation is not fully automated — and it shouldn’t be. Certain decisions require human judgment: Is this model response harmful? Does this evaluation rubric make sense for this domain? Is this edge case a genuine failure or acceptable behavior? The workflow orchestrator manages these decision points. It coordinates multi-step evaluation workflows where some steps are automated (LLM scoring, data validation) and others require human approval before the pipeline continues.

AI 评估并非完全自动化，也不应该完全自动化。某些决策需要人类的判断：这个模型的回答是否有害？这个评估准则对该领域是否合理？这个边缘案例是真正的故障还是可接受的行为？工作流编排器负责管理这些决策点。它协调多步骤的评估工作流，其中一些步骤是自动化的（如 LLM 评分、数据验证），而另一些步骤则需要在流水线继续之前获得人工批准。

The Architecture

架构

The core is a state machine built on FastAPI and PostgreSQL. Each workflow is a DAG (directed acyclic graph) of tasks, where each node can be:

  • Automated: Runs immediately, calls another service (scoring, data enrichment), stores the result
  • Human gate: Pauses the workflow, notifies the assigned reviewer via the notification service, waits for approval/rejection
  • Conditional: Routes to different branches based on previous step outcomes (e.g., if confidence score < threshold, escalate to senior reviewer)

其核心是一个基于 FastAPI 和 PostgreSQL 构建的状态机。每个工作流都是一个任务的 DAG（有向无环图），其中每个节点可以是：

  • 自动化节点： 立即运行，调用其他服务（评分、数据增强），并存储结果。
  • 人工关卡： 暂停工作流，通过通知服务提醒指定的审核员，并等待批准或拒绝。
  • 条件节点： 根据前一步的结果路由到不同的分支（例如，如果置信度分数低于阈值，则升级给高级审核员）。

State transitions are persisted in PostgreSQL with Alembic-managed migrations. Every transition is logged — who approved what, when, and with what context. This audit trail turned out to be critical for client reporting.

状态转换通过 Alembic 管理的迁移持久化存储在 PostgreSQL 中。每一次转换都会被记录下来——谁在何时、在什么背景下批准了什么。事实证明，这种审计追踪对于客户报告至关重要。

Real-Time Updates with WebSocket

使用 WebSocket 实现实时更新

The original system polled the API every 5 seconds to check workflow status. With dozens of reviewers working concurrently, this created unnecessary load and a poor user experience — you’d approve a task and see nothing happen for up to 5 seconds. I replaced this with WebSocket connections that push state changes in real-time. When a reviewer approves a step, every connected client watching that workflow sees the update instantly. The implementation uses FastAPI’s WebSocket support with Redis Pub/Sub as the message broker, so it works across multiple Cloud Run instances.

最初的系统每 5 秒轮询一次 API 以检查工作流状态。当数十名审核员同时工作时，这造成了不必要的负载和糟糕的用户体验——你批准了一个任务，却要等上 5 秒才能看到变化。我将其替换为实时推送状态变化的 WebSocket 连接。当审核员批准某个步骤时，所有正在查看该工作流的连接客户端都会立即看到更新。该实现使用了 FastAPI 的 WebSocket 支持，并以 Redis Pub/Sub 作为消息代理，因此它可以在多个 Cloud Run 实例之间正常工作。

# Simplified WebSocket broadcast pattern
# 简化的 WebSocket 广播模式
async def broadcast_workflow_update(workflow_id: str, event: dict):
    channel = f"workflow:{workflow_id}"
    await redis.publish(channel, json.dumps({
        "type": "state_change",
        "workflow_id": workflow_id,
        "step": event["step"],
        "status": event["new_status"],
        "actor": event["actor_email"],
        "timestamp": datetime.utcnow().isoformat()
    }))

Production Logging Overhaul

生产环境日志重构

The existing codebase used print() statements everywhere. In production on Cloud Run, these were effectively invisible — they’d show up as unstructured text in Cloud Logging with no way to filter, search, or correlate them. I replaced the entire logging infrastructure with structured JSON logging. Every log entry includes a correlation ID that traces a request across the workflow orchestrator, the notification service, and whatever downstream service is involved. When a workflow fails at step 4 of 7, you can now trace exactly what happened at each step, in each service, with a single query.

现有的代码库到处都在使用 print() 语句。在 Cloud Run 的生产环境中，这些语句实际上是不可见的——它们在 Cloud Logging 中显示为非结构化文本，无法进行过滤、搜索或关联。我用结构化的 JSON 日志替换了整个日志基础设施。每个日志条目都包含一个关联 ID，用于追踪请求在工作流编排器、通知服务以及任何涉及的下游服务中的流转。当工作流在 7 个步骤中的第 4 步失败时，你现在可以通过单个查询准确追踪到每个步骤、每个服务中发生了什么。

Taxonomy Workflow Engine: Intelligent Task Classification

分类法工作流引擎：智能任务分类

Not all evaluation tasks are the same. A code generation task requires different rubrics, different evaluators, and different tooling than a conversational AI task. The taxonomy engine is the routing layer that classifies incoming tasks and determines which evaluation workflow to apply.

并非所有的评估任务都是一样的。代码生成任务与对话式 AI 任务相比，需要不同的准则、不同的评估者和不同的工具。分类法引擎是路由层，负责对传入的任务进行分类，并确定应用哪种评估工作流。

The Problem It Solves

它解决的问题

Before this service existed, task classification was manual. A project manager would look at incoming evaluation requests, decide which team should handle them, and assign the appropriate rubric. This worked at 50 tasks per day. It didn’t work at thousands.

在该服务存在之前，任务分类是手动的。项目经理需要查看传入的评估请求，决定由哪个团队处理，并分配相应的准则。这在每天 50 个任务时还能应付，但在数千个任务时就失效了。

How It Works

工作原理

The engine uses a combination of keyword matching, metadata analysis, and configurable rule sets to classify tasks. Each classification determines:

  • Which evaluation rubric to apply
  • Which reviewer pool to draw from (by expertise)
  • Whether the task requires single or multi-reviewer consensus
  • SLA targets for completion time

该引擎结合了关键词匹配、元数据分析和可配置的规则集来对任务进行分类。每次分类决定了：

  • 应用哪种评估准则
  • 从哪个审核员池中抽取（按专业领域）
  • 任务是否需要单人或多人共识
  • 完成时间的 SLA 目标

The file upload system allows clients to submit evaluation tasks in bulk via CSV/JSON uploads to GCS. The engine parses, validates, classifies each row, and enqueues them into the appropriate workflow — all asynchronously via Cloud Tasks.

文件上传系统允许客户通过 CSV/JSON 上传到 GCS 来批量提交评估任务。引擎会对每一行进行解析、验证、分类，并将其加入到相应的工作流中——所有这些都通过 Cloud Tasks 异步完成。

Infrastructure: Cloud SQL for taxonomy rules and classification history, GCS for bulk file uploads, Cloud Run for the API layer, Cloud Tasks for async processing.

基础设施： 用于存储分类规则和分类历史的 Cloud SQL，用于批量文件上传的 GCS，用于 API 层的 Cloud Run，以及用于异步处理的 Cloud Tasks。

Core Platform Service: The Authentication Backbone

核心平台服务：身份验证骨干

Every microservice in the platform needs to answer two questions: “Who is making this request?” and “Are they allowed to do this?” The core platform service provides those answers.

平台中的每个微服务都需要回答两个问题：“谁在发起此请求？”以及“他们被允许这样做吗？”核心平台服务提供了这些答案。

JWT Authentication Fixes

JWT 身份验证修复

The existing JWT implementation had a subtle but critical bug: token validation was checking expiration time against the server’s local time rather than UTC. Cloud Run instances can have slight clock drift, and this meant tokens would occasionally be rejected as “expired” when they were still valid, or accepted when they should have been rejected. The fix was straightforward — normalize all time comparisons to UTC — but finding it required tracing sporadic 401 errors across multiple services to realize the pattern correlated with specific Cloud Run instances, not specific users.

现有的 JWT 实现有一个微妙但关键的错误：令牌验证是将过期时间与服务器的本地时间而不是 UTC 时间进行比较。Cloud Run 实例可能会有轻微的时钟漂移，这意味着令牌有时会在仍然有效时被拒绝为“已过期”，或者在应该被拒绝时被接受。修复方法很简单——将所有时间比较标准化为 UTC——但要发现这个问题，需要跨多个服务追踪零星的 401 错误，才能意识到这种模式与特定的 Cloud Run 实例相关，而不是与特定的用户相关。

# Before: clock-sensitive comparison
# 之前：对时钟敏感的比较
if token_exp < datetime.now(): # Local time — unreliable on Cloud Run
    raise HTTPException(401, "Token expired")

# After: UTC-normalized comparison
# 之后：标准化为 UTC 的比较
if token_exp < datetime.now(timezone.utc):
← 返回新闻列表 / Back to news list