> ## Content Index
> Fetch the complete content index at: https://www.hubwiz.com/blog/llms.txt
> Use this file to discover other available public pages before exploring further.

# 我的多智能体数据管道
- URL: https://www.hubwiz.com/blog/my-multi-agent-data-pipeline/
- Published: 2026-03-06T07:54:17.000Z
- Updated: 2026-03-06T07:54:17.000Z
- Description: 我构建了多个AI智能体来自动化大约 80% 的数据管道胶水工作。不是完全自主。不是炒作。只是围绕无聊、可重复任务的紧密范围的自动化。
- Author: admin
- Tags: AGENT, #Import 2026-10-14 14:41

一个新的 CSV 会落在 S3 中。我会下载它，扫描缺失的列，将其与我们的规范模式进行比较，编写一个快速转换脚本，健全性检查聚合，然后为利益相关者起草一个简短摘要。这些都不是智力困难的。只是重复的。到了周四，我会为不同的来源再次这样做。

工作并不复杂。它是令人疲惫的。认知成本不在于编写代码——而在于一遍又一遍地做出相同的小决策。

最终，我停止优化 SQL 查询，开始优化决策循环。

这是可重现的多代理 AI 工作流程演练，我构建了它来自动化大约 80% 的数据管道胶水工作。不是完全自主。不是炒作。只是围绕无聊、可重复任务的紧密范围的自动化。

## 1、问题：人类胶水工作

在自动化之前，我的每周管道是这样的：

- 验证新数据（模式漂移、必需列）
- 将新字段映射到规范模式
- 建议简单的转换
- 运行异常检查
- 编写简短摘要报告

每周花费时间：**\~10 小时**。

不是因为系统脆弱。因为人类充当了胶水。

消耗不是技术困难。它是：

- 重新评估模式不匹配
- 决定 `total_amt` 是否映射到 `total_amount`
- 重新检查 17% 的激增是合法的还是损坏的
- 重复编写类似的报告

这是一种结构化的、可重复的决策制定，代理系统实际上有意义的地方。

## 2、架构：小代理，清晰边界

我没有构建"超级代理"。我构建了具有窄责任的小代理。

1. **摄取验证代理**
2. **模式映射代理**
3. **异常检测代理**
4. **报告代理**

没有代理拥有工作流程。

每一个：

- 接受结构化输入
- 产生严格验证的 JSON
- 具有单一责任

ASCII 图表：多代理流程

```
[Raw Data]
    |
[Validator Agent]
    |
[Schema Agent]
    |
[Anomaly Agent]
    |
[Report Agent]

```

每个阶段都是确定性的。编排层决定顺序和重试。

这不是自主混乱。它是结构化流程。

## 3、编排器：无混乱的控制

编排器故意无聊。

它处理：

- 排序
- 重试
- 模式验证
- 升级

这里是 Python 中的简化版本：

```python
def run_pipeline(data):
    validated = validator.run(data)
    mapped = schema_agent.run(validated)
    anomalies = anomaly_agent.run(mapped)
    report = reporter.run(anomalies)

    return report

```

在生产中，每个 `.run()`：

- 验证 JSON 模式
- 应用重试策略
- 记录提示 + 响应
- 跟踪延迟

编排器从不委派控制。代理不互相调用。所有交接都是明确的。

### 代理 1：摄取验证器

责任：

- 检查必需列
- 检测模式漂移
- 标记格式错误的记录

它接收：

```json
{
  "columns": ["order_id", "total_amt", "created_at"],
  "sample_rows": [...]
}

```

结构化提示风格：

```
prompt = f"""
You are a data ingestion validator.

Required columns: {required_columns}
Incoming columns: {incoming_columns}

Return JSON:
{{
  "missing_columns": [],
  "unexpected_columns": [],
  "is_valid": true|false
}}
"""

```

使用严格模式解析：

```python
from pydantic import BaseModel

class ValidationResult(BaseModel):
    missing_columns: list[str]
    unexpected_columns: list[str]
    is_valid: bool

response = llm.invoke(prompt)
result = ValidationResult.model_validate_json(response)

```

如果 `is_valid` 为 false，管道停止。

没有猜测。没有自动修复。

### 代理 2：模式映射代理

责任：

- 将传入列映射到规范模式
- 建议转换

示例输出：

```json
{
  "mappings": {
    "total_amt": "total_amount"
  },
  "transformations": {
    "total_amount": "float(value)"
  },
  "confidence": 0.92
}

```

护栏：

- 规范模式是固定的并注入
- 代理不能发明新的规范字段
- 置信度 < 0.8 触发人工审查

示例调用：

```python
mapping = schema_agent.run({
    "incoming_columns": incoming_columns,
    "canonical_schema": canonical_schema
})

if mapping.confidence < 0.8:
    escalate_to_human(mapping)

```

这是幻觉早期发生的地方。护栏修复了它。

### 代理 3：异常检测代理

这个代理有两层：

1. 统计预检查（代码）
2. 推理层（LLM）

首先，确定性阈值：

```python
def precheck(metrics):
    alerts = []
    if metrics["daily_revenue_change"] > 0.25:
        alerts.append("Revenue spike > 25%")
    return alerts

```

然后上下文推理：

```
prompt = f"""
Metrics:
{metrics}

Existing alerts:
{alerts}

Explain whether anomalies appear legitimate.
Return JSON:
{{
  "summary": "",
  "severity": "low|medium|high",
  "confidence": 0-1
}}
"""

```

回退行为：

```python
result = anomaly_agent.run(data)

if result.confidence < 0.7:
    result.severity = "needs_review"

```

代理从不单独决定。它在确定性检查之上分层推理。

## 4、故障处理和护栏

每个边界验证输出。

#### ASCII 图表：护栏循环

```
[Agent Output]
     |
  Validate
     |
   Retry / Escalate

```

策略：

- 格式错误的 JSON 自动重试 2 次
- 严格的 Pydantic 验证
- 低置信度升级
- 版本化提示
- 记录输入 + 输出

这将脆弱的提示变成了受控组件。

## 5、之前 vs 之后：真正的影响

这是部署此 AI 数据管道自动化后发生的变化：

- 手动时间：**10 小时/周 → 2 小时/周**
- 模式映射错误：**\~12% → \~3%**
- 审查周期：从 3 轮减少到 1 轮
- 吞吐量：每月上线的数据源约多 2 倍

最大的收获不是速度。它是一致性。

系统每次都以相同的方式做出相同的决策。

## 6、碰到了哪些问题

这并不顺利。

早期失败：

- 幻觉规范字段
- 过于自信的映射
- 列名奇怪时的提示脆弱性
- JSON 格式漂移

修复：

- 显式模式注入
- 置信度评分
- 严格的 JSON 解析
- 确定性预检查层
- 代理之间的硬边界

只有当我们像不可靠的微服务一样对待代理时，系统才变得稳定。

## 7、为什么这对管道有效

管道是结构化的。

它们有：

- 已知的模式
- 重复的任务
- 可验证的输出
- 确定性阶段

这使它们成为多代理 AI 工作流的理想选择。

当任务模糊且无界时，代理系统失败。当自主权受到限制且输出可验证时，它们成功。

这是应用于无聊决策的可重现 AI 工作流程代码。

## 8、这会失败的地方

这种方法在以下情况下崩溃：

- 高歧义领域（法律解释）
- 有责任的高度合规工作流程
- 超低延迟系统
- 实时事务处理

它不适用于创造性推理或开放式判断。

它是用于结构化胶水工作。

## 9、代理的可观测性

没有可观测性的代理变得无法调试。

#### ASCII 图表：可观测性层

```
[Agent] -> [Logs]
        -> [Metrics]
        -> [Trace]

```

我们跟踪什么：

- 提示版本
- 输入负载哈希
- 输出 JSON
- 每个代理的延迟
- 重试计数
- 升级率

我们还分类错误：

- 解析失败
- 低置信度标志
- 漂移检测不匹配

这使系统可维护。

## 10、使其工作的设计原则

三件事很重要：

1. **小代理**
2. **确定性编排**
3. **在每个边界验证**

没有代理拥有管道。

没有隐式链接。

没有静默失败。

这不是关于 LLM 代理后端炒作。它是关于减少结构化系统中的重复决策制定。

### 11、*最终*反思

工作流程并不消除工程师。

它消除了工程师必须第五十次决定 `total_amt` 可能意味着 `total_amount` 的部分。

这里最强大的自动化不是智能。

它减少人类必须做出相同决策两次的次数。

---

原文链接: [How I Built a Multi-Agent AI Workflow That Automates 80% of My Data Pipeline](https://medium.com/@yashbatra11111/how-i-built-a-multi-agent-ai-workflow-that-automates-80-of-my-data-pipeline-746078e8ac71?ref=hubwiz.com)

汇智网翻译整理，转载请标明出处