制造业 AI 落地架构：别急着上 LangChain，先建好"企业数据能力层"

做 AI 交付久了，我见过最多的一种场景：一个制造工厂找来说，我要上 AI，API 接个 DeepSeek，再套个 LangChain Agent，完事。

然后呢？数据孤岛依然在，MES 的数据 ERP 不认识，CRM 叫的 SKU 在车间叫物料编码，AI 一问三不知——不是模型不行，是底座没搭。

这篇文章，不讲 Demo，不画饼。直接给一套分层的、可落地的企业 AI 架构，以及为什么你应该在接触 LangChain 之前，先把"数据能力层"建好。

📎 相关阅读： 《制造业 AI 不是替代岗位，是吃掉岗位里的低价值脑力劳动》 — 本文讲技术架构怎么搭；那篇讲这套架构对每个岗位意味着什么，哪些工作被自动化，哪些工作永远不会被替代。两篇建议配合阅读。

一、不要直接接 LangChain，先做"企业数据能力层"

一个典型的制造工厂，数据来源通常是这样的：

MES — 生产制造执行系统（工单、工序、设备、良率）
ERP — 企业资源计划（采购、库存、财务）
CRM — 客户关系管理（客户、合同、需求）
PLM — 产品生命周期管理（产品数据、BOM、图纸版本）
WMS — 仓储管理系统（入库、出库、库位、批次）
IoT — 工业物联网（设备传感器、温度/压力/振动、实时数据流）
文件服务器 — 工艺文件、SOP、质量标准、设备图纸
OA/钉钉 — 审批流程、项目文档、沟通记录

最大的问题不是没有数据，而是数据孤岛。各系统各自为政，同样的东西叫不同名字，且彼此互不相通。

所以第一步——也是最容易被跳过的一步——是建设企业数据接入层：

MES

ERP

CRM

PLM

WMS

IoT

文件

钉钉

│ │ │ │ │

▼ ▼ ▼ ▼ ▼

企业数据接入层
API / 数据库直连 / ETL / 文档解析

▼

统一数据底座
PostgreSQL / ClickHouse / 向量库

结构化数据接入

MES / ERP / CRM 等业务系统的结构化数据，通过以下方式接入：

API — 系统开放接口（RESTful / GraphQL）
数据库直连 — 只读从库，定时同步 CDC（Change Data Capture）
ETL 管道 — 定时批处理，数据清洗与标准化

典型的 MES 生产表结构示例：

订单号	产品	工序	设备	良率	时间
MO-20260701	注塑件-A100	工序3	机台5	95%	14:30

统一接入目标：PostgreSQL（事务查询）或 ClickHouse（分析查询），构建企业级数据仓库。

技术选型决策：PostgreSQL vs ClickHouse

选 PostgreSQL 当：

订单查询、工单状态、权限事务——需要实时写入和行级更新，数据量在 TB 以下

选 ClickHouse 当：

设备振动秒级指标、良率趋势分析、供应链聚合——场景以宽表聚合为主，数据量 10TB+

经验法则：事务场景用 PG，分析场景用 CK。如果规模不大（单表 <5 亿行），PG + 物化视图基本够用，没必要为了大数据而上 ClickHouse。

非结构化数据接入

PDF、Word、Excel、工艺图纸、SOP 等非结构化文档，接入方案完全不同：

文档解析 — PDF 解析、OCR 识别、表格提取
切分与向量化 — Chunking + Embedding
存入向量数据库 — 构建 RAG 知识库底座

二、建立企业知识层（RAG / LlamaIndex）

数据接入之后，不要急着把原始数据丢给 Agent。需要分类处理：

结构化知识 → 数据库查询

这类知识有明确的字段定义和关系结构，适合通过 SQL 或 API 直接查询：

订单A → 生产线1 → 工序3 → 异常率5%

不需要向量检索，一个 JOIN 就能解决。

非结构化知识 → RAG 检索

《注塑工艺规范》、《质量异常处理流程》这类文档，适合通过 RAG 检索：

📄 文件（PDF / Word / Excel / 图纸）

▼

🔍 文档解析（PDF解析 / OCR / 表格提取）

▼

✂️ 文本切片（Chunking）

▼

🧬 Embedding 向量化

▼

🗄 向量数据库
Milvus / pgvector / Chroma

▼

🔎 语义检索
LlamaIndex / LangChain Retriever

推荐技术栈：LlamaIndex 做文档管理和索引，Milvus 或 pgvector 做向量存储，LangChain Retriever 做检索接入。

技术选型决策：pgvector vs Milvus vs Chroma

pgvector

文档数 <100 万，不想维护额外组件。嵌入 PostgreSQL 同一集群，备份简单，适合起步。

Milvus

文档数 100 万+，需要分布式检索和高吞吐。适合二次检索和长文档库的正式生产环境。

Chroma

开发原型、Demo、POC 阶段。一行代码跑起来，但不要在生产环境持久化使用。

三、做业务知识建模（Ontology / 知识建模）

这一步非常关键，但很多团队会忽略。制造业尤其需要本体建模，因为不同系统间的叫法太乱了。

一个真实例子：

MES 里的 产品编码 = "100-A"
ERP 里的 物料编码 = "M001"
CRM 里的 SKU = "A100"

这三个字段，AI 看到后不知道它们指向的是同一个东西。你期待 AI 能推理"订单延期原因"，但它连哪条数据对应什么都分不清。

本体做什么？

本体（Ontology）解决的就是这个问题——建立一个企业业务世界模型，让 AI 理解这个企业的数据世界是什么样的。

核心实体（Entity）

产品  →  客户  →  订单  →  物料  →  设备  →  工序  →  员工  →  供应商
                

关系（Relation）

客户

下达 →

销售订单

生成 →

生产订单

使用 →

物料

生产订单

经过 →

工序

使用 →

设备

产生 →

质量数据

属性（Attribute）

例如产品实体：

产品: { 编号, 型号, 规格, 客户 }
设备: { 编号, 状态, 位置, 维护周期 }
质量问题: { 缺陷类型, 严重等级, 原因, 解决方案 }
                

有本体 vs 无本体

没有本体时：

老板问"为什么客户 A 的订单延期？" Agent 只会说"我去查一下订单表"——但不知道订单关联哪条生产线、哪个设备、哪个物料、哪个供应商。

有本体后：

Agent 理解这条链路：

客户A → 订单B → 产品C → 工单D → 工序3 → 设备E → 异常记录F
                

然后自动规划：

查订单状态
查生产进度
查设备运行记录
查质量异常
查物料供应情况

这就是知识驱动 Agent——不是黑盒推理，而是有路径、可追溯、可解释的推理。

Ontology + RAG 的融合架构

💬 用户问题

▼

🧠 本体理解问题中的实体与关系

▼

📌 定位相关数据源和文档

▼

📚 RAG 补充知识细节

▼

🤖 Agent 推理与决策

举个例子：问"为什么这个客户最近交付总是延期？"

本体识别出链路：

客户 → 订单 → 产品 → 工序 → 设备

RAG 检索到《设备异常处理规范》，发现 3 号线的某台设备连续两次停机导致工序 3 延迟。Agent 综合输出："因设备 X 故障导致，建议按 SOP 更换核心模块。"

四、把企业能力封装成 Tool / Skill

完成数据和知识层面的基建之后，才进入 Agent 层。

原则：不要让 Agent 直接访问数据库。每一层的能力都需要封装为独立的 Tool。

Tool 清单示例

get_production_status(order_id)

查询当前生产订单状态、完成率、异常情况

get_quality_issue(product_id)

查询产品质量异常和历史处理方案

get_inventory(material_id)

查询物料库存和供应商交期

get_customer_history(customer_id)

查询客户历史订单和交付记录

query_process_sop(keyword)

检索工艺规范 / SOP 文件

Agent 看到的就是这些能力：生产查询、质量分析、库存查询、客户分析、工艺知识检索。

至于背后是 MES 数据库还是 ERP 接口，Agent 不需要关心。这是标准的适配器模式。

真实 Tool 实现：用 Python 写一个生产查询 Tool

以下是一个对接真实 MES 数据库的生产查询 Tool 实现，展示了从原始 SQL 查询到 Agent 可调用的 Function 的完整链路：

import json
from typing import Optional
from pydantic import BaseModel, Field

# ---- 1. Tool Schema（LangChain Tool 签名） ----

class ProductionStatusInput(BaseModel):
    order_id: str = Field(description="生产订单号，如 MO-20260701")
    user_role: str = Field(default="admin", description="调用者角色")

# ---- 2. 数据层：对接 MES 数据库 ----

import psycopg2
from psycopg2.extras import RealDictCursor

DB_CONFIG = {
    "host": "10.3.0.10", "port": 5432,
    "dbname": "mes_production",
    "user": "mes_readonly", "password": "***",
}

def _query_mes(sql: str, params: dict) -> list[dict]:
    """只读从库查询"""
    conn = psycopg2.connect(**DB_CONFIG)
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(sql, params)
            return [dict(row) for row in cur.fetchall()]
    finally:
        conn.close()

# ---- 3. 权限层：行级数据过滤 ----

ROLE_FILTERS = {
    "admin":    "",
    "manager":  "AND line_id IN (SELECT line_id FROM user_lines WHERE user_id = %(user_id)s)",
    "operator": "AND station_id = (SELECT station_id FROM users WHERE user_id = %(user_id)s)",
}

# ---- 4. Tool 实现 ----

def get_production_status(order_id: str, user_role: str = "admin") -> str:
    """查询生产订单状态、完成率、异常情况"""
    filter_sql = ROLE_FILTERS.get(user_role, "")
    sql = f"""
        SELECT o.order_id, o.product_name, o.quantity,
               o.completed_qty, o.status,
               (SELECT COUNT(*) FROM quality_anomalies
                WHERE order_id = o.order_id AND severity = 'high'
               ) AS critical_anomalies
        FROM production_orders o
        WHERE o.order_id = %(order_id)s {filter_sql}
    """
    rows = _query_mes(sql, {"order_id": order_id})
    if not rows:
        return json.dumps({"error": "订单不存在或无权访问"})
    row = rows[0]
    rate = round(row["completed_qty"] / row["quantity"] * 100, 1)
    return json.dumps({
        "order_id": row["order_id"],
        "product": row["product_name"],
        "status": row["status"],
        "completion_rate": f"{rate}%",
        "critical_anomalies": row["critical_anomalies"],
    }, ensure_ascii=False)

# ---- 5. 注册为 LangChain Tool ----

from langchain_core.tools import StructuredTool

production_tool = StructuredTool.from_function(
    func=get_production_status,
    name="get_production_status",
    description="查询生产订单状态、完成率和异常情况",
    args_schema=ProductionStatusInput,
)
                

关键设计要点：

只读从库 — Tool 只连接只读副本，绝不写生产库
权限内嵌 SQL — user_role 参数在 SQL 层直接过滤行级数据
统一返回 JSON — 所有 Tool 统一 JSON 字符串，方便 Agent 解析和组合
Schema 标准化 — Pydantic 定义输入，LangChain 自动生成 Function Calling 定义

五、企业 Agent 接入标准：MCP

Tool 封装解决了单个能力的服务化问题，但企业迟早要面对一个新问题：每个新系统（MES/ERP/WMS/CRM…）都需要写一套自定义的 Tool，怎么做到一次接入、到处可用？

MCP（Model Context Protocol） 是答案。MCP 是 Anthropic 提出的一种开放协议，它定义了 Agent ↔ 企业数据源之间的标准通信方式。理解 MCP 的最佳方式是类比 USB-C 接口——不管后面接的是显示器、硬盘还是扩展坞，前面都是同一个口子。

🧠 LangChain / LangGraph Agent

▼

🔌 MCP Client（标准接口）

┌────┼────┴────┼────┐

MCP Server
MES

MCP Server
ERP

MCP Server
WMS

MCP Server
PLM

MCP Server
IoT

MCP 的核心价值

协议化接入，不是代码化接入 — 不再需要为每个业务系统专门写一个 LangChain Tool。新系统上线时，只需部署一个 MCP Server（几十行 Python），集团 Agent 零代码接入。
动态发现能力 — MCP Server 可以实时宣告自己有哪些 Tool（Resources + Tools），Agent 端无需预先注册。
安全边界清晰 — MCP Client 和 Server 之间是标准化的认证/授权机制，可以在企业网络边界统一管控。
多 Agent 复用 — 同一个 MCP Server（如 MES API）可以被生产 Agent、质量 Agent、交付 Agent 同时调用，不重复开发 Tool。

一个 MCP Server 的典型实现（Python，使用 mcp 库）：

from mcp import Server, tool

server = Server("mes-production")

@tool()
async def get_production_status(order_id: str):
    """查询生产订单状态"""
    # 内部调用 MES 接口
    return await mes_client.query(order_id)

@tool()
async def get_equipment_status(equipment_id: str):
    """查询设备运行状态"""
    return await equipment_client.query(equipment_id)

if __name__ == "__main__":
    server.run("stdio")  # 支持 stdio 和 SSE 两种传输模式
                

MCP 在企业架构中的位置

在之前的分层架构中，MCP 位于「Tool 封装层」之上——它不是替代 Tool，而是给 Tool 提供一个标准化的对外接口：

🧠 Agent Orchestration
LangGraph / LangChain

▼

🔌 MCP Protocol Layer

┌────────┼────────┐

🛠 Tool
MES/ERP/WMS

🛠 Tool
PLM API

🛠 Tool
IoT 数据流

└───┴───┘

▼

🗄 企业数据源

MCP vs. 直接 Tool 封装：什么时候选哪个

场景	推荐方式	理由
单个 Agent、单一系统	直接 Tool 封装	简单直接，复杂度最低
多个 Agent 共享同一数据源	MCP	一次接入，多 Agent 复用
新系统频繁接入	MCP	标准化接入，新人/新系统上手快
安全审计要求高	MCP	边界统一的认证/授权机制
仅仅做一个 Demo 验证	直接 Tool	先跑通，不要过度工程设计

最后说一点：MCP 不是银弹。如果你的数据层（一、二章）本身一团糟，上了 MCP 也只是把混乱标准化了而已。理清数据、建好知识层，MCP 是在这个基础上的锦上添花。

六、给 Agent 加上权限体系

工具封装好了，但还有一个关键问题没解决：谁可以调用什么？

制造企业的数据天然有层级。老板应该能看到全部数据，生产经理只看得了自己那条产线，产线工人只能看到自己的工位。如果不做权限控制，一个 Agent 的返回结果对车间主任和 CEO 是一样的——这显然不合理。

正确的设计是：权限层嵌入在 Tool 调用链中。

👤 用户身份

▼

🔐 权限过滤器
角色 → 可访问范围

▼

🔧 Tool 调用
携带权限上下文

▼

📊 数据返回
仅返回权限范围内的数据

权限设计示例

👑 老板

全部数据：所有产线、所有订单、质量报告、财务数据

🏭 生产经理

本产线数据：工单状态、设备运行、异常记录、良率统计

🔧 产线员工

本工位数据：当班工单、操作指引、自检记录

两种实现方式

方式一：Tool 层面过滤（推荐）

每个 Tool 在定义时就接受 user_role 参数，在数据查询阶段直接拼接权限过滤条件：

def get_production_status(order_id, user_role):
    if user_role == 'operator':
        # 只返回该员工所在工位的数据
        return query("SELECT * FROM production WHERE station = :station", station=user.station)
    elif user_role == 'manager':
        # 返回该经理管辖产线的数据
        return query("SELECT * FROM production WHERE line_id = :line", line=user.line)
    elif user_role == 'admin':
        # 返回全部数据
        return query("SELECT * FROM production")
                

方式二：数据层统一过滤（更彻底）

在数据库/API 网关层实施行级安全（RLS），Tool 层无需关心权限逻辑，数据层自动根据用户身份过滤：

-- PostgreSQL Row Level Security
CREATE POLICY production_policy ON production
    USING (
        current_user = 'admin' OR
        (current_user = 'manager' AND line_id IN (SELECT line_id FROM user_lines WHERE user_id = current_user_id())) OR
        (current_user = 'operator' AND station_id = current_station())
    );
                

对企业来说，推荐方式一：Tool 层权限过滤。原因有三：

Agent 的 Tool 本身就是抽象层，天然适合在此做权限控制
不需要改造底层数据库（很多老旧 MES 根本没 RLS 能力）
权限变更时可以只改 Tool 逻辑，不影响数据层

但方式二更适合新建系统或有能力改造数据层的场景。两种也可以混合使用：数据层做基础行级隔离，Tool 层做业务层权限增强。

七、LangChain 负责 Agent 编排

到了这一步，LangChain 才正式登场——作为 Agent 编排层，负责任务的拆解与执行。

👤 用户

▼

LangChain Agent (ReAct)
思考 → 行动 → 观察 → 推理

┌──────┼──────┐──────┐

Tool1
MES

Tool2
ERP

Tool3
WMS

RAG
知识库

Memory
历史

实际场景：老板问"为什么订单 A 延期？"

调用 MES Tool：查询订单 A 的生产进度
发现：工序 3 异常率上升
调用质量分析 Tool：查该异常的历史处理方案
调用 RAG：查相关 SOP 和工艺规范
生成报告：延期原因 + 解决方案 + 恢复建议

八、复杂场景建议用 LangGraph

LangChain 的 Agent 适合简单任务。制造业很多场景是有状态的流程，这时候应该上 LangGraph。

因为制造流程天然是状态驱动的：

🚀 开始

▼

📋 查询订单

▼

❓ 是否延期？

▼

⚙️ 查询生产进度

▼

❓ 判断质量

▼

📄 生成报告

▼

✋ 人工审批

▼

✅ 结束

这不是聊天，是流程。有明确的节点、分支、状态转换和人工介入点。LangGraph 的有向图（DAG）模型天然适合这类场景。

LangGraph 节点定义示例

下面用 LangGraph 实现上面的订单延期诊断流程，展示 state 如何在节点间传递：

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END

# ---- 1. 定义 State（贯穿全流程的上下文） ----

class OrderState(TypedDict):
    order_id: str
    user_role: str
    status: str | None
    completion_rate: float | None
    is_delayed: bool | None
    anomalies: list | None
    report: str | None

# ---- 2. 定义节点（函数） ----

def fetch_order(state: OrderState) -> dict:
    """节点 A：查询订单"""
    result = get_production_status(state["order_id"], state["user_role"])
    data = json.loads(result)
    rate = float(data.get("completion_rate", "0").replace("%", ""))
    return {"status": data["status"], "completion_rate": rate}

def check_delay(state: OrderState) -> dict:
    """节点 B：判断是否延期"""
    delayed = state["completion_rate"] < 80 or state["status"] == "delayed"
    return {"is_delayed": delayed}

def diagnose(state: OrderState) -> dict:
    """节点 C：诊断质量问题"""
    issues = get_quality_issue(state["order_id"])
    return {"anomalies": json.loads(issues)}

def generate_report(state: OrderState) -> dict:
    """节点 D：生成报告"""
    r = f"订单 {state['order_id']} 诊断报告\n"
    r += f"- 状态：{state['status']}\n"
    r += f"- 完成率：{state['completion_rate']}%\n"
    if state.get("anomalies"):
        r += f"- 异常：{state['anomalies'][:3]}\n"
    r += "- 建议：按 SOP 检查设备"
    return {"report": r}

def approve(state: OrderState) -> dict:
    """节点 E：人工审批"""
    return {"report": state["report"] + "\n（待人工审批）"}

# ---- 3. 条件边（路由器） ----

def router(state: OrderState) -> str:
    if state["is_delayed"]:
        return "diagnose"
    return "generate_report"

# ---- 4. 构建图 ----

builder = StateGraph(OrderState)
builder.add_node("fetch", fetch_order)
builder.add_node("check", check_delay)
builder.add_node("diagnose", diagnose)
builder.add_node("gen_report", generate_report)
builder.add_node("approval", approve)

builder.set_entry_point("fetch")
builder.add_edge("fetch", "check")
builder.add_conditional_edges("check", router)
builder.add_edge("diagnose", "gen_report")
builder.add_edge("gen_report", "approval")
builder.add_edge("approval", END)
builder.add_edge("gen_report", END)

app = builder.compile()

# ---- 5. 执行 ----

result = app.invoke({
    "order_id": "MO-20260701",
    "user_role": "admin",
})
print(result["report"])
                

对比 LangChain Agent：所有逻辑挤在 ReAct 循环里，状态存内存、分支靠提示词、人工审批几乎不可控。而 LangGraph 把每个节点、每个分支、每个状态转换都显式定义了——可维护性、可追踪性、可测试性完全是两个档次。

九、Agent 运行治理层（AgentOps / 可观测性）

前面八章讲的都是 Agent 怎么做，但企业真正决定上线前一定会问：AI 出错怎么办？

生产环境不能只靠「再跑一次」——每一次错误都是真实成本。正确的做法是在 Agent 和 Tool 之间加一层 Agent 运行治理层（Agent Runtime），把每一次运行都记录、追踪、评分。

🧠 LangChain / LangGraph Agent

▼

⚙️ Agent 运行治理层
Agent Runtime / Governance

┌────┼────┼────┼────┼────┐

📝 Prompt 版本管理

📋 Tool 调用日志

💰 Token 成本

⭐ 答案质量评分

👤 人工反馈闭环

└───┴───┴───┴───┴───┘

▼

🔌 MCP / Tool 层

治理层核心功能

1. Prompt 版本管理

生产环境中 Prompt 不可能一成不变。业务方今天说「加个规则」，明天 QA 说「某个案例错了」。没有版本管理的话，每次改 Prompt 都是提心吊胆——因为不知道改了什么，也不知道改了之后影响什么。

做法：每次修改 Prompt 必须走版本化流程：写新版本 → A/B 测试 → 灰度上线 → 全量切换。LangSmith 的 Hub 或自定义 Git + YAML 都可以。

# prompt-history.yaml
prompts:
  production_agent_v1:
    content: "你是生产管理专家，查询生产信息，给出分析。"
    created: 2026-03-01
  production_agent_v2:
    content: "你是生产管理专家，查询生产信息，分析异常原因，给出建议。"
    created: 2026-04-15
    diff: "增加了分析异常原因和建议"
  production_agent_v3:
    content: "你是生产管理专家，查询生产信息，分析异常原因给出建议，如果异常连续出现需标记红线。"
    created: 2026-05-20
    diff: "增加红线标记逻辑"
                

2. Tool 调用日志

Agent 每次调用 Tool 的参数、返回值、耗时都必须记录。这是排查 Agent 行为的唯一手段——没有日志就没有可观测性。

建议日志结构：

run_id — 每次对话的唯一 ID，关联所有 Trace
tool_name — 调用了哪个 Tool
input_params — 传入参数（注意脱敏）
output — 返回值摘要
duration_ms — 调用耗时
success — 是否成功
error_message — 失败原因

LangSmith 自动完成这些，但如果企业要求数据不出网，可以用 OpenTelemetry 自搭。

3. Token 成本追踪

企业上 Agent 最怕的除了出错就是超预算。每个对话的 Token 消耗必须可量化。

追踪维度：

每次调用 — 输入 Token + 输出 Token
每笔成本 — 按模型定价换算
每日/周/月汇总 — 按 Agent、部门、场景分类
异常告警 — 单次成本超过阈值时自动告警

LangSmith 提供开箱即用的成本追踪面板。Arize Phoenix 也支持 Token 用量监控。

4. 答案质量评分

Agent 回答得好不好不能只靠感觉，需要可量化的质量标准。

自动评分 — 用 LLM-as-Judge 对答案质量打分（相关性、准确性、完整性）
用户评分 — 每次回复提供「有用/没用」按钮，收集用户反馈
回归测试 — 维护评测集，每次改 Prompt 或换模型自动跑一遍，看分数是否下降

5. 人工反馈闭环

这是治理层最重要但也最容易被忽略的一环。Agent 不可能百分之百正确，关键是出错了之后走什么流程。

闭环设计：

Agent 回答 → 用户标记「不对」
记录上下文 + Agent 推理链 + Tool 调用链
推送到人工审核队列
审核员给出正确答案 / 修复方案
修复结果反哺到评测集，防止相同错误再次出现

十、最终架构图

👤 用户

▼

📱 企业 AI 助手（飞书 / 钉钉 / Web）

▼

🧠 Agent 编排层
LangGraph / LangChain

▼

⚙️ Agent 运行治理层
Trace · Logging · Cost · Eval · Feedback

└───┘

▼

🔌 MCP 协议层
标准化 Agent ↔ 企业数据接口

┌────┼────┼────┐

🛠 Tool 封装
MES/ERP/WMS/PLM

📚 RAG 知识检索
IoT + 向量 + 图谱

💾 Memory
历史上下文

└───┴───┴───┘

▼

🔧 数据治理层
接入 / 清洗 / 本体建模 / 权限

▼

MES

ERP

CRM

PLM

WMS

IoT

文件

钉钉

这个架构的核心思想是分层：数据治理层统一接入企业数据源；知识层建模业务语义；MCP 协议层标准化接口；Tool/RAG/Memory 各司其职；治理层保障 Trace、成本和质量；Agent 层只做编排，不关心底层实现。

架构搭好了，然后呢？

这套分层架构的本质是自动化企业里的信息处理工作。
它对每个岗位意味着什么？哪些人最需要，哪些人最抗拒？
哪些工作被彻底自动化，哪些反而变得更值钱？

→ 读《制造业 AI 不是替代岗位，
是吃掉岗位里的低价值脑力劳动》

十一、第一阶段不要做"大一统 Agent"

制造企业做 AI 最容易犯的错就是："做一个万能 AI 助手"。这个想法很诱人，但大概率死在路上。

建议先做三个垂直 Agent。至于每个 Agent 对应哪些岗位、自动化了哪些工作、对人的影响是什么，可以配合阅读 《制造业 AI 不是替代岗位》 ——那篇从岗位维度拆解得非常细。

1. 生产运营 Agent

用户：生产经理
能力：查订单状态、查产线产能、查异常工单

2. 质量 Agent

用户：质量经理
能力：查不良数据、查缺陷原因、查 SOP 规范

3. 销售交付 Agent

用户：业务经理
能力：查客户信息、查订单进度、查预计交期

每个 Agent 只做一件事，做好了再合并、交叉、扩展。

十二、技术架构全景

🤖 Agent

▼

LangGraph

┌────────┼────────┐

Ontology
Neo4j

RAG
Milvus

Tools
MES/ERP

│ │ │

知识图谱

向量库

业务系统

└───┴───┘

▼

🗄 企业数据源
MES · ERP · CRM · PLM · WMS · IoT · 文件 · 钉钉

关键组件：

Ontology 存储：Neo4j 或其他图数据库
RAG 向量库：Milvus / pgvector
本体建模工具：Protégé 或自定义 RDF/OWL
Agent 编排：LangGraph / LangChain
结构化数据：PostgreSQL / ClickHouse
企业接入：飞书 / 钉钉 Bot

十三、推荐落地路线

第一阶段：RAG + Tool

先解决"能查、能问、能执行"的问题。集成数据、封装 Tool、搭建基础 RAG 知识库。

第二阶段：业务本体

建立企业核心实体的映射关系，让 AI 理解"M001 = A100 = 100-A"这种跨系统语义对齐。

第三阶段：GraphRAG

结合知识图谱与向量检索，实现复杂的多跳推理——"设备停机导致工序延迟，进而影响订单交付"这类推理链。

不要一开始就做"完整知识图谱"。很多项目死在这里。本体是逐步演化的，不是一次建成的。

制造业天然适合本体——产品结构（BOM）、工艺路线、设备关系、供应链关系、质量追溯——这些本质就是知识图谱。但做得太大、太快，反而会变成一把干不完的活。

渐进式推进，每次只解决一个痛点，才是制造企业 AI 落地的正确姿势。

十四、常见失败模式与避坑指南

上面的架构看起来很完整，但实际交付中，每一层都可能翻车。以下是我在各种项目里见过最多的死法：

1. 数据接入层：Schema 变更导致管道断裂

MES 系统的表结构三天两头变——Excel 导入模板改了、字段名变了、某列从数字变成枚举了。如果没有 schema 版本管理，ETL 管道会静默吞掉数据而不报错，AI 查到的数据其实是残缺的。

对策：ETL 链路加 schema 校验和告警，检测到字段名或类型变化自动暂停管道并通知。用 Great Expectations 做数据质量断言，每次 ETL 跑完自动生成质量报告。

2. 知识层：RAG 召回噪声导致 Agent 幻觉

非结构化文档质量参差不齐——有的 SOP 是 2019 年的旧版本，有的 PDF 扫描件 OCR 识别率只有 60%。Agent 一旦被低质量召回内容误导，输出比没有知识更危险——它会自信地编造一个错误答案。

对策：对召回结果加相关性打分阈值，低于 0.7 的不返回 Agent。给文档加元数据（版本号、部门、生效日期），回答中标注来源和置信度。最关键的一条：如果 Agent 找不到相关知识，让它明确说不知道，不要强行回答。

3. 本体层：本体膨胀后维护成本爆炸

一开始只建了 10 个实体，半年后膨胀到 200 个。实体间的关系定义、属性更新、跨系统映射校准全部靠人工维护，全公司没人能说清楚本体图长什么样。最后成了一个没人敢动的怪物。

对策：本体只建到够用为止。先覆盖核心业务（产品-订单-物料-工序），其他按需扩展。本体变更走审批流程和回归测试。定期清理无人使用的实体和关系。

4. Tool 层：Agent 链路长超时和级联失败

一个订单延期诊断涉及 5 个 Tool 调用，其中 MES 数据库查询需要 5 秒。整个推理链被一个慢查询卡死。更糟糕的是，某个内部 API 挂了，Agent 不会优雅降级，直接抛异常退出。

对策：每个 Tool 设超时（建议 3-5 秒），超时后返回服务暂不可用而不是抛异常。关键 Tool 做降级——MES 查不到时至少返回订单 ID 存在但详细信息暂不可用，而不是直接报错打断推理。

5. 权限层：权限遗漏导致数据泄露

新增了一个 Tool，开发忘了加 user_role 过滤，结果车间工人通过 Agent 能查到全公司的订单和财务数据。这不是假设，这是真实发生过的事。

对策：Tool 注册时强制要求声明最低权限角色，没有权限声明的 Tool 不允许上线。写一个 e2e 测试脚本，用 admin / manager / operator 三种角色身份跑同一个查询，断言输出范围符合预期。

常见问题（FAQ）

Q：一定要用 Neo4j 吗？能不能用关系数据库存本体？

可以用。实体数 <500 时关系数据库完全够用。Neo4j 的优势在于多跳查询——设备停机会影响哪些订单，进而影响哪些客户？SQL 的 JOIN 链会越来越长，Cypher 天生支持路径遍历。从关系表起步完全 OK，后期迁移到图库也不复杂。

Q：LangChain 版本迭代太快，上生产不放心怎么办？

锁定大版本，用 langchain==0.3.x 不要追最新版。LangChain 的核心是 Tool 抽象和 ReAct 循环，这两个 API 在大版本内是稳定的。如果确实担心，可以绕过 LangChain 自己实现 Tool 调度——核心代码不到 200 行。

Q：做 RAG 应该用哪种 Embedding 模型？

中文场景推荐 BAAI/bge-large-zh-v1.5 或 text2vec-large-chinese，在 MTEB 中文榜上表现稳定。中英混排文档用 multilingual-e5-large。不建议用 OpenAI 的 text-embedding-3 做中文 RAG——不是效果不行，而是延迟和成本在制造企业的文档量级下撑不住。

Q：制造企业 AI 团队至少要什么配置？

最小配置：1 个后端（懂 Python + SQL），1 个业务分析（懂制造流程），1 个项目经理。AI 算法能力可以外采 API（DeepSeek / Qwen），不需要自训模型。关键是非技术能力——懂业务的那个人需要能画清楚产品→订单→工序→设备的链路图。

Q：你这套架构和低代码 AI 平台（Dify / Coze）冲突吗？

不冲突。低代码平台可以替代上层的 Agent 编排和 RAG 搭建。但下三层的基建——数据接入、本体建模、Tool 封装——无论用什么平台都绕不开。这篇文章讲的就是把下三层做好，上层用什么工具挂上去反而可以灵活选择。