标签：AI Agent OA集成 MES系统 LangChain RAG 企业数智化

---

一、前言：为什么企业AI必须接入OA/MES？

在企业数智化转型中，AI不能只当"聊天机器人"，必须真正参与业务流程。OA承载审批、请假、报销等；MES管控生产排程、设备状态、质量检测等。

传统RPA强依赖DOM树/XPath，系统升级就失效。新一代AI Agent通过API编排或非侵入式视觉理解，实现真正的"像人一样操作"。

本文基于 LangChain + RAG + 多Agent架构，手把手教你搭建同时对接OA和MES的AI智能体。

---

二、整体架构设计

2.1 架构全景图

用户交互层 (Web/钉钉/语音)
         |
         v
   AI Agent 核心层
   - 意图识别(Intent Classifier)
   - OA Agent / MES Agent / 通用问答 / 数据分析
   - 记忆系统(Memory): Milvus向量库 + 衰减权重
   - 工具调用(Tool Use): OA API / MES API / DB查询
         |
         v
   数据与知识层
   - OA知识库 / MES知识库 / 企业制度文档 (RAG索引)
   - OA数据库(MySQL) / MES数据库(SQL Server) / Milvus向量库

2.2 四层核心架构

层级	职责	关键技术
感知层	接收用户输入，多模态处理	NLP意图识别、语音识别
推理层	大模型理解需求，任务拆解	GPT-4/Claude + 知识图谱
执行层	调用外部API，执行操作	API编排、Function Calling
反馈层	收集结果，持续优化	在线学习、RLHF

---

三、环境准备与项目搭建

3.1 技术栈选型

• 后端：Python 3.10 + FastAPI

• AI框架：LangChain + LangGraph

• 大模型：GPT-4 / 通义千问 / 文心一言

• 向量库：Milvus

• 数据库：MySQL(OA) + SQL Server(MES)

• 缓存：Redis

• 部署：Docker + Docker Compose

3.2 项目目录

ai-enterprise-agent/
├── agent_core/          # Agent核心
│   ├── intent_classifier.py
│   ├── oa_agent.py
│   ├── mes_agent.py
│   ├── memory_manager.py
│   └── tool_registry.py
├── rag_system/          # RAG知识检索
│   ├── document_loader.py
│   ├── vector_store.py
│   └── retriever.py
├── tools/               # 外部工具
│   ├── oa_tools.py
│   ├── mes_tools.py
│   └── db_tools.py
├── api/
│   └── main.py          # FastAPI入口
├── docker-compose.yml
└── requirements.txt

3.3 依赖安装

# requirements.txt
langchain>=0.2.0
langchain-openai>=0.1.0
langgraph>=0.0.50
pymilvus>=2.4.0
fastapi>=0.110.0
uvicorn>=0.27.0
sqlalchemy>=2.0.0
pyodbc>=5.0.0
requests>=2.31.0
redis>=5.0.0
python-dotenv>=1.0.0

---

四、核心模块实操

4.1 意图识别模块 - 系统的"交通指挥官"

为什么需要意图识别？

当用户说"帮我查一下昨天产线3的报警"，系统需要判断这是MES查询；当用户说"帮我提交请假申请"，系统需要路由到OA Agent。如果没有意图识别，所有请求都走同一个Agent，会导致工具调用混乱、回答质量下降。

核心思路： 用GPT-4o做分类器，temperature设为0.1保证输出稳定。用Enum定义6种意图类型，让分类结果可枚举、可追踪。

# agent_core/intent_classifier.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from enum import Enum
​
class IntentType(Enum):
    OA_QUERY = "oa_query"       # OA查询类：查审批、查公告
    OA_ACTION = "oa_action"     # OA操作类：提交申请、审批通过
    MES_QUERY = "mes_query"     # MES查询类：查设备、查报警
    MES_ACTION = "mes_action"   # MES操作类：提交维修、调整参数
    GENERAL_CHAT = "general_chat"   # 通用对话
    DATA_ANALYSIS = "data_analysis" # 数据分析
​
class IntentClassifier:
    def __init__(self):
        # temperature=0.1 让模型输出更稳定，适合做分类任务
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
        
    def classify(self, user_input: str) -> IntentType:
        # 用ChatPromptTemplate构建分类提示词
        # system消息定义分类规则，human消息传入用户输入
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a router. Classify user intent into: oa_query, oa_action, mes_query, mes_action, data_analysis, general_chat. OA involves leave, reimbursement, approval. MES involves production line, device, work order, alarm."),
            ("human", "Input: {input}")
        ])
        # 用管道操作符 | 构建执行链：prompt -> llm
        chain = prompt | self.llm
        response = chain.invoke({"input": user_input})
        # 清洗输出：转小写、去空格
        intent_str = response.content.strip().lower()
        try:
            return IntentType(intent_str)
        except ValueError:
            # 如果模型输出不在Enum中，默认走通用对话
            return IntentType.GENERAL_CHAT
​
# 使用示例
classifier = IntentClassifier()
print(classifier.classify("帮我查昨天产线3的报警"))  # -> mes_query
print(classifier.classify("帮我提交请假申请"))         # -> oa_action

关键点：

• temperature=0.1 分类任务不需要创意，越低越稳定

• 用 Enum 而不是字符串，避免拼写错误导致的路由失败

• 异常兜底 ValueError 捕获，确保系统不会崩溃

---

4.2 OA Agent - 对接办公系统

OA Agent 做什么？

OA Agent 是专精办公自动化场景的"专家Agent"。它只持有OA相关的工具（查审批、提交请假、查公告），不会误调用MES的工具。这种"专精分工"是多Agent架构的核心优势。

核心思路： 用 create_openai_tools_agent 创建ReAct风格的Agent，让大模型自主决定：先查什么、再查什么、最后怎么组织回答。

# agent_core/oa_agent.py
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from tools.oa_tools import OATools
​
class OAAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
        # 从OATools获取所有OA相关工具
        self.tools = OATools().get_tools()
        
        # 构建Agent提示词模板
        # system: 定义Agent身份和能力边界
        # chat_history: 对话历史占位符（支持多轮对话）
        # input: 用户当前输入
        # agent_scratchpad: Agent思考过程的"草稿纸"
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an OA assistant. Handle approval queries, leave applications, announcements. Sensitive operations require identity confirmation."),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        
        # 创建OpenAI Tools Agent（支持Function Calling）
        self.agent = create_openai_tools_agent(self.llm, self.tools, self.prompt)
        # AgentExecutor负责执行：思考 -> 调用工具 -> 观察结果 -> 再思考 -> ...
        self.executor = AgentExecutor(agent=self.agent, tools=self.tools, verbose=True)
    
    def run(self, user_input: str, chat_history=None):
        return self.executor.invoke({"input": user_input, "chat_history": chat_history or []})

# tools/oa_tools.py
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
import requests
from typing import Optional
​
# 用Pydantic定义每个工具的输入参数，LLM会自动根据参数描述生成调用
class QueryApprovalInput(BaseModel):
    approval_type: str = Field(description="leave/salary/purchase")
    status: Optional[str] = Field(default=None, description="pending/approved/rejected")
    date_range: Optional[str] = Field(default=None, description="e.g. 2024-07-01~2024-07-31")
​
class SubmitLeaveInput(BaseModel):
    leave_type: str = Field(description="annual/sick/personal")
    start_date: str = Field(description="start date")
    end_date: str = Field(description="end date")
    reason: str = Field(description="reason")
​
class OATools:
    def __init__(self, base_url="http://oa.company.com/api"):
        self.base_url = base_url
        self.headers = {"Authorization": "Bearer {token}"}
    
    def _query_approval(self, approval_type, status=None, date_range=None):
        # Query approval records
        params = {"type": approval_type}
        if status: params["status"] = status
        if date_range: params["date_range"] = date_range
        try:
            resp = requests.get(f"{self.base_url}/approvals", params=params, timeout=10)
            data = resp.json()
            if not data.get("records"): return "No records found"
            records = data["records"]
            result = f"Found {len(records)} records:\n"
            for r in records[:5]:  # 最多返回5条，避免消息过长
                result += f"- [{r['status']}] {r['title']} | {r['submitter']} | {r['create_time']}\n"
            return result
        except Exception as e:
            return f"Query failed: {str(e)}"
    
    def _submit_leave(self, leave_type, start_date, end_date, reason):
        # Submit leave application
        payload = {
            "type": "leave",
            "leave_type": leave_type,
            "start_date": start_date,
            "end_date": end_date,
            "reason": reason
        }
        try:
            resp = requests.post(f"{self.base_url}/approvals", json=payload, timeout=10)
            data = resp.json()
            if data.get("success"):
                return f"Success! Approval ID: {data['approval_id']}"
            return f"Failed: {data.get('message', 'unknown error')}"
        except Exception as e:
            return f"Submit failed: {str(e)}"
    
    def get_tools(self):
        # Register all tools for Agent use
        return [
            StructuredTool.from_function(
                func=self._query_approval,
                name="query_approval",
                description="Query OA approval records",
                args_schema=QueryApprovalInput  # Pydantic model constrains parameter format
            ),
            StructuredTool.from_function(
                func=self._submit_leave,
                name="submit_leave",
                description="Submit leave application",
                args_schema=SubmitLeaveInput
            ),
        ]

关键点：

• StructuredTool + Pydantic 让LLM能准确理解每个参数的含义和格式

• args_schema 是LangChain的"魔法"——LLM根据Field的description自动填参数

• timeout=10 防止OA系统卡死导致Agent hang住

• 异常处理返回字符串而不是抛异常，Agent能继续推理

---

4.3 MES Agent - 对接制造执行系统

MES Agent 和 OA Agent 的区别？

MES Agent 面对的是实时数据和工业安全规范。设备状态每秒都在变，报警信息需要即时响应，而且任何误操作都可能导致产线停机。所以MES Agent的system prompt里特别强调"二次确认"和"人工复核"。

# agent_core/mes_agent.py
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from tools.mes_tools import MESTools
​
class MESAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
        self.tools = MESTools().get_tools()
        
        # system prompt emphasizes safety rules - key difference from OA
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a MES assistant. Query device status, OEE, work orders, alarms. Safety: device operations require double confirmation."),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        self.agent = create_openai_tools_agent(self.llm, self.tools, self.prompt)
        self.executor = AgentExecutor(agent=self.agent, tools=self.tools, verbose=True)
    
    def run(self, user_input, chat_history=None):
        return self.executor.invoke({"input": user_input, "chat_history": chat_history or []})

# tools/mes_tools.py
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
import pyodbc  # SQL Server driver
from datetime import datetime, timedelta
from typing import Optional
​
class QueryDeviceStatusInput(BaseModel):
    line_id: str = Field(description="e.g. LINE-01")
    device_id: Optional[str] = Field(default=None)
​
class QueryAlarmInput(BaseModel):
    line_id: Optional[str] = Field(default=None)
    level: Optional[str] = Field(default=None, description="critical/warning/info")
    time_range: str = Field(default="24h", description="1h/24h/7d")
​
class MESTools:
    def __init__(self):
        # MES typically uses SQL Server
        self.db_config = {
            "server": "192.168.1.100",
            "database": "MES_DB",
            "username": "mes_user",
            "password": "mes_password"
        }
    
    def _get_db_connection(self):
        # Encapsulate DB connection for reuse
        conn_str = (
            f"DRIVER={{ODBC Driver 17 for SQL Server}};"
            f"SERVER={self.db_config['server']};"
            f"DATABASE={self.db_config['database']};"
            f"UID={self.db_config['username']};"
            f"PWD={self.db_config['password']}"
        )
        return pyodbc.connect(conn_str)
    
    def _query_device_status(self, line_id, device_id=None):
        # Query device real-time status
        # Why connect directly to DB instead of API?
        # MES real-time data is usually in SQL Server or time-series DB.
        # Direct DB query is faster than API layer, especially for high-frequency queries.
        try:
            conn = self._get_db_connection()
            cursor = conn.cursor()
            if device_id:
                # Query single device
                sql = """SELECT device_name, status, oee, runtime, downtime
                         FROM device_status WHERE line_id=? AND device_id=?"""
                cursor.execute(sql, (line_id, device_id))
            else:
                # Query all devices in line
                sql = """SELECT device_name, status, oee, runtime, downtime
                         FROM device_status WHERE line_id=? ORDER BY device_id"""
                cursor.execute(sql, (line_id,))
            rows = cursor.fetchall()
            conn.close()
            
            if not rows:
                return f"No devices found for line {line_id}"
            
            # Use emoji for intuitive output
            result = f"Line {line_id} Status:\n" + "-"*40 + "\n"
            for row in rows:
                emoji = "🟢" if row.status=="running" else "🔴" if row.status=="down" else "🟡"
                result += f"{emoji} {row.device_name} | {row.status} | OEE:{row.oee}%\n"
            return result
        except Exception as e:
            return f"Query failed: {str(e)}"
    
    def _query_alarms(self, line_id=None, level=None, time_range="24h"):
        # Query alarm records
        # time_range uses dict mapping to support natural language input like 1h/24h/7d.
        # Automatically calculates start time, avoiding complex time formats for users.
        try:
            # Natural language to hours
            hours = {"1h": 1, "24h": 24, "7d": 168}.get(time_range, 24)
            start_time = datetime.now() - timedelta(hours=hours)
            
            conn = self._get_db_connection()
            cursor = conn.cursor()
            
            # Dynamic SQL, only append filters with values
            sql = """SELECT alarm_time, line_id, device_name, alarm_level, alarm_msg, status
                     FROM alarm_log WHERE alarm_time>=?"""
            params = [start_time]
            
            if line_id:
                sql += " AND line_id=?"
                params.append(line_id)
            if level:
                sql += " AND alarm_level=?"
                params.append(level)
            sql += " ORDER BY alarm_time DESC"
            
            cursor.execute(sql, params)
            rows = cursor.fetchall()
            conn.close()
            
            if not rows:
                return f"No alarms in past {time_range}"
            
            result = f"Alarms ({len(rows)} total):\n" + "-"*40 + "\n"
            for row in rows[:10]:  # Limit return count to prevent message overflow
                emoji = {"critical":"🔴", "warning":"🟠", "info":"🔵"}.get(row.alarm_level, "⚪")
                result += f"{emoji} [{row.alarm_level.upper()}] {row.alarm_time}\n"
                result += f"   Line:{row.line_id} | Device:{row.device_name}\n"
                result += f"   {row.alarm_msg} | Status:{row.status}\n\n"
            return result
        except Exception as e:
            return f"Query failed: {str(e)}"
    
    def get_tools(self):
        return [
            StructuredTool.from_function(
                func=self._query_device_status,
                name="query_device_status",
                description="Query device real-time status",
                args_schema=QueryDeviceStatusInput
            ),
            StructuredTool.from_function(
                func=self._query_alarms,
                name="query_alarms",
                description="Query alarm records",
                args_schema=QueryAlarmInput
            ),
        ]

关键点：

• MES直接连SQL Server而不是REST API，因为工业数据查询频率高、实时性强

• pyodbc 是Python连接SQL Server的标准方案，注意安装 ODBC Driver 17

• 动态SQL拼接避免None值污染查询条件

• emoji增强可读性，运维人员一眼看出设备状态

---

4.4 RAG知识库 - 让AI懂企业制度

为什么需要RAG？

大模型训练数据截止到某个时间点，不可能知道你们公司的请假制度、MES操作规范。RAG（检索增强生成）的作用就是：把企业私有文档（PDF、Word、网页）变成向量，存进Milvus，用户提问时先检索相关知识，再让大模型基于检索结果回答。

核心流程： 文档加载 -> 文本切分 -> 向量化 -> 存入Milvus -> 查询时向量检索 -> 拼接上下文 -> LLM生成回答

# rag_system/vector_store.py
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
​
class EnterpriseRAG:
    def __init__(self, milvus_host="192.168.184.128", milvus_port="19530"):
        # Connect to Milvus vector database
        # Using your previous config: 192.168.184.128:19530
        connections.connect(alias="default", host=milvus_host, port=milvus_port)
        
        # OpenAI text-embedding-3-large, 3072 dims, better than ada-002
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.collection_name = "enterprise_knowledge"
        self._init_collection()
    
    def _init_collection(self):
        # Initialize Milvus collection (table)
        # Load if exists, create if not.
        # Schema: id(auto PK) + content + source + category + embedding(vector)
        if utility.has_collection(self.collection_name):
            self.collection = Collection(self.collection_name)
            return
        
        fields = [
            FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
            FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512),
            FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=64),
            # text-embedding-3-large outputs 3072-dim vectors
            FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=3072)
        ]
        schema = CollectionSchema(fields, "Enterprise Knowledge")
        self.collection = Collection(self.collection_name, schema)
        
        # IVF_FLAT index: suitable for millions of records, balance of speed and accuracy
        self.collection.create_index("embedding", {
            "index_type": "IVF_FLAT",
            "metric_type": "L2",      # Euclidean distance, good for semantic similarity
            "params": {"nlist": 128}   # Cluster centers, larger for bigger datasets
        })
        self.collection.load()
    
    def add_documents(self, documents, category="general"):
        # Add documents to knowledge base
        # Why split into chunks?
        # LLMs have context length limits (4k-128k). Long text vectors dilute key info.
        # Split into ~500 char chunks, each independently vectorized for precise retrieval.
        # RecursiveCharacterTextSplitter: split by priority recursively
        # First by double newline, then single newline, then period, preserving semantic integrity
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,      # Max 500 chars per chunk
            chunk_overlap=50,    # 50 char overlap between adjacent chunks, prevent info loss at boundaries
            separators=["\n\n", "\n", ".", ";", " "]
        )
        
        chunks, sources, categories = [], [], []
        for doc in documents:
            splits = splitter.split_text(doc["content"])
            chunks.extend(splits)
            sources.extend([doc["source"]] * len(splits))
            categories.extend([category] * len(splits))
        
        # Batch vector generation, much faster than one-by-one
        embeddings = self.embeddings.embed_documents(chunks)
        
        # Insert into Milvus. Note: entity order must match field definition
        self.collection.insert([chunks, sources, categories, embeddings])
        self.collection.flush()  # Ensure data is persisted
    
    def search(self, query, category=None, top_k=5):
        # Vector search
        # Flow: query -> vectorize -> find top_k most similar chunks in Milvus -> return content+source
        # category parameter filters, e.g. only search MES-related docs
        query_embedding = self.embeddings.embed_query(query)
        
        # If category specified, add filter to prevent OA docs from interfering with MES queries
        expr = f'category == "{category}"' if category else None
        
        results = self.collection.search(
            data=[query_embedding],           # Query vector
            anns_field="embedding",           # Field to search on
            param={"metric_type": "L2", "params": {"nprobe": 10}},
            limit=top_k,                      # Return top_k results
            expr=expr,                        # Filter condition
            output_fields=["content", "source"]  # Fields to return
        )
        
        return [
            {
                "content": hit.entity.get("content"),
                "source": hit.entity.get("source"),
                "score": hit.distance  # L2 distance, smaller = more similar
            }
            for hit in results[0]
        ]
​
# ========== Usage Example ==========
rag = EnterpriseRAG()
​
# Add OA policy docs, mark category="oa"
rag.add_documents([
    {"content": "Leave requires 3 days advance notice. Annual leave must be approved by direct manager and HR.", "source": "Employee Handbook"}
], category="oa")
​
# Add MES operation standards, mark category="mes"
rag.add_documents([
    {"content": "Critical alarms require immediate shutdown and notify maintenance team. Do not restart until root cause is identified.", "source": "MES SOP"}
], category="mes")
​
# Search with category for precise retrieval
results = rag.search("What to do when device alarms?", category="mes")
for r in results:
    print(f"[{r['source']}] {r['content'][:100]}...")

关键点：

• chunk_overlap=50 防止关键信息被切分在边界处丢失

• category 字段实现"知识隔离"，OA文档不会干扰MES查询

• IVF_FLAT 索引适合百万级以下数据，如果数据量更大可以换 HNSW

• nprobe=10 控制搜索精度，越大越准但越慢

---

4.5 记忆系统 - 让对话有上下文

为什么需要记忆？

没有记忆的AI就像金鱼——每轮对话都从零开始。用户上一句问"产线3的报警"，下一句问"那设备状态呢？"，AI需要知道"那"指的是产线3。

核心思路： 把对话历史也存进Milvus，用向量检索找到"语义相关"的历史记录（不只是时间最近的），让AI真正"记得"你们聊过什么。

# agent_core/memory_manager.py
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility
from datetime import datetime
​
class ConversationMemory:
    def __init__(self, milvus_host="192.168.184.128", milvus_port="19530"):
        # Use separate alias="memory" to avoid connection conflicts with RAG
        connections.connect(alias="memory", host=milvus_host, port=milvus_port)
        self.collection_name = "conversation_memory"
        self._init_collection()
    
    def _init_collection(self):
        if utility.has_collection(self.collection_name):
            self.collection = Collection(self.collection_name)
            return
        
        fields = [
            FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name="session_id", dtype=DataType.VARCHAR, max_length=64),
            FieldSchema(name="timestamp", dtype=DataType.INT64),  # Unix timestamp for sorting
            FieldSchema(name="role", dtype=DataType.VARCHAR, max_length=16),  # user/assistant
            FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
            FieldSchema(name="intent", dtype=DataType.VARCHAR, max_length=32),
            FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=3072)
        ]
        schema = CollectionSchema(fields, "Conversation Memory")
        self.collection = Collection(self.collection_name, schema)
        self.collection.create_index("embedding", {
            "index_type": "IVF_FLAT",
            "metric_type": "L2",
            "params": {"nlist": 64}
        })
        self.collection.load()
    
    def add_message(self, session_id, role, content, intent=None):
        # Add a conversation record
        # Each sentence is vectorized and stored in Milvus.
        # Later, semantic retrieval can find related content.
        # E.g. user asks "How is that device?", vector search finds previous device discussions.
        from langchain.embeddings import OpenAIEmbeddings
        emb = OpenAIEmbeddings(model="text-embedding-3-large")
        embedding = emb.embed_query(content)
        
        ts = int(datetime.now().timestamp())
        # Note: insert params are 2D lists, one list per field
        self.collection.insert([
            [session_id],      # session_id
            [ts],              # timestamp
            [role],            # role
            [content],         # content
            [intent or "unknown"],  # intent
            [embedding]        # embedding
        ])
        self.collection.flush()
    
    def search_relevant(self, session_id, query, top_k=3):
        # Semantic search for relevant historical memory
        # Not simply taking the most recent N entries, but finding the N most semantically relevant.
        # E.g. user previously asked about "Line 3 CNC device", now asks "Is that device fixed?",
        # vector search can connect these two sentences even across many conversation turns.
        from langchain.embeddings import OpenAIEmbeddings
        emb = OpenAIEmbeddings(model="text-embedding-3-large")
        query_emb = emb.embed_query(query)
        
        # Only search current session's memory to prevent cross-contamination
        results = self.collection.search(
            data=[query_emb],
            anns_field="embedding",
            param={"metric_type": "L2", "params": {"nprobe": 10}},
            limit=top_k,
            expr=f'session_id == "{session_id}"',  # Key: isolate different users' memories
            output_fields=["role", "content", "timestamp"]
        )
        
        return [
            {
                "role": hit.entity.get("role"),
                "content": hit.entity.get("content")
            }
            for hit in results[0]
        ]

关键点：

• 用 session_id 隔离不同用户/对话的记忆，避免A用户看到B用户的内容

• 语义检索比时间排序更智能，能找回"相关但久远"的记忆

• 记忆和RAG用同一个Milvus实例但不同collection，资源复用

---

五、主入口与路由编排

主入口做什么？

这是整个系统的"总调度中心"，每轮对话的执行流程：

1. 检索相关记忆 -> 2. 检索RAG知识 -> 3. 识别意图 -> 4. 路由到对应Agent -> 5. 保存对话

# api/main.py
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional
from agent_core.intent_classifier import IntentClassifier, IntentType
from agent_core.oa_agent import OAAgent
from agent_core.mes_agent import MESAgent
from agent_core.memory_manager import ConversationMemory
from rag_system.vector_store import EnterpriseRAG
from langchain_openai import ChatOpenAI
import uuid
​
app = FastAPI(title="Enterprise AI Agent", version="1.0.0")
​
# Global initialization (production: use dependency injection)
classifier = IntentClassifier()
oa_agent = OAAgent()
mes_agent = MESAgent()
memory = ConversationMemory()
rag = EnterpriseRAG()
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
​
class ChatRequest(BaseModel):
    message: str
    session_id: Optional[str] = None  # If empty, create new session
    user_id: Optional[str] = None
​
class ChatResponse(BaseModel):
    response: str
    intent: str
    sources: Optional[List[str]] = None  # Return cited knowledge sources
​
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    # Generate new UUID if no session_id provided
    session_id = request.session_id or str(uuid.uuid4())
    user_input = request.message
    
    # ========== Step 1: Retrieve relevant memory ==========
    # Search history with current query to find semantically related context
    relevant_memories = memory.search_relevant(session_id, user_input)
    memory_context = "\n".join([f"{m['role']}: {m['content']}" for m in relevant_memories])
    
    # ========== Step 2: Retrieve RAG knowledge ==========
    # No category specified, let system match based on query content
    # (Can also dynamically specify based on intent)
    rag_results = rag.search(user_input, top_k=3)
    rag_context = "\n".join([f"[{r['source']}] {r['content']}" for r in rag_results])
    
    # ========== Step 3: Intent classification ==========
    intent = classifier.classify(user_input)
    
    # ========== Step 4: Route to corresponding Agent ==========
    # Inject memory and RAG knowledge into prompt for context-aware answers
    if intent in [IntentType.OA_QUERY, IntentType.OA_ACTION]:
        enriched = f"User: {user_input}\n\nMemory:\n{memory_context}\n\nKnowledge:\n{rag_context}"
        result = oa_agent.run(enriched)
        response_text = result["output"]
    elif intent in [IntentType.MES_QUERY, IntentType.MES_ACTION]:
        enriched = f"User: {user_input}\n\nMemory:\n{memory_context}\n\nKnowledge:\n{rag_context}"
        result = mes_agent.run(enriched)
        response_text = result["output"]
    elif intent == IntentType.DATA_ANALYSIS:
        # Data analysis can use dedicated Agent or direct LLM
        response_text = f"Data analysis in development. Intent: {intent.value}"
    else:
        # General chat: use LLM directly, inject memory and knowledge for better quality
        prompt = f"You are an enterprise assistant.\nUser: {user_input}\nMemory: {memory_context}\nKnowledge: {rag_context}"
        response_text = llm.invoke(prompt).content
    
    # ========== Step 5: Save conversation ==========
    # Store user input and AI response for future retrieval
    memory.add_message(session_id, "user", user_input, intent.value)
    memory.add_message(session_id, "assistant", response_text, intent.value)
    
    return ChatResponse(
        response=response_text,
        intent=intent.value,
        sources=[r["source"] for r in rag_results] if rag_results else None
    )
​
@app.get("/health")
async def health_check():
    return {"status": "ok", "version": "1.0.0"}
​
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

关键点：

• session_id 是记忆隔离的关键，Web端通常存在localStorage，钉钉/企微可以用用户ID

• 记忆和RAG知识都拼进prompt，但放在system prompt之后，避免干扰Agent的system指令

• sources 返回给用户，增加可信度（"根据《员工手册》第3条..."）

---

六、Docker部署

# docker-compose.yml
version: '3.8'
services:
  ai-agent:
    build: .
    ports: ["8000:8000"]
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - MILVUS_HOST=milvus-standalone
      - MILVUS_PORT=19530
    depends_on: [milvus-standalone, redis]
    networks: [agent-network]
​
  milvus-standalone:
    image: milvusdb/milvus:v2.4.0
    ports: ["19530:19530", "9091:9091"]
    volumes: [milvus_data:/var/lib/milvus]
    environment:
      - ETCD_ENDPOINTS=etcd:2379
      - MINIO_ADDRESS=minio:9000
    networks: [agent-network]
​
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
    networks: [agent-network]
​
  minio:
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    networks: [agent-network]
​
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    networks: [agent-network]
​
volumes:
  milvus_data:
networks:
  agent-network:
    driver: bridge

部署说明：

• depends_on 只保证容器启动顺序，不保证服务就绪，生产环境建议加健康检查

• Milvus的etcd和minio是内部依赖，不需要暴露端口到宿主机

• Redis用于缓存和异步任务队列（Celery），本例中未展示但建议预留

---

七、应用场景演示

场景1：OA智能助手

用户：帮我查一下我的请假审批进度
​
AI：Found 2 records:
      - [approved] Annual leave (7.15-7.19) | 2024-07-10
      - [pending] Sick leave (7.20) | 2024-07-18
      Annual leave approved, sick leave pending manager review.

流程解析：

1. 意图识别：classify() -> oa_query

2. 路由到 OAAgent

3. Agent决策：调用 query_approval 工具，参数 approval_type="leave"

4. 工具执行HTTP请求，返回JSON

5. Agent整理成自然语言回复

场景2：MES产线监控

用户：产线3昨天有什么报警？
​
AI：Line LINE-03 alarms (3 total):
      🔴 [CRITICAL] 14:23 CNC-03 Spindle temp exceeded (85C) | Resolved
      🟠 [WARNING] 09:15 ROBOT-01 Fixture pressure low | Resolved
      🟠 [WARNING] 16:45 CONVEYOR-02 Speed fluctuation | Monitoring
      Suggestion: Check CNC-03 cooling system.

流程解析：

1. 意图识别：classify() -> mes_query

2. 路由到 MESAgent

3. Agent调用 query_alarms，参数 line_id="LINE-03", time_range="24h"

4. 工具直连SQL Server查询 alarm_log 表

5. Agent用emoji和格式化输出增强可读性

场景3：跨系统联动

用户：产线2设备故障了，帮我提交维修申请
​
AI：Cross-system operation:
      Step 1: LINE-02 CONVEYOR-01 status: down
      Step 2: Submit maintenance request...
      Success! Ticket: MR-20240718-0032
      Priority: Urgent | ETA: 30 minutes
      Maintenance team and line supervisor notified.

流程解析：

1. 意图识别：classify() -> mes_action（先查MES设备状态）

2. MESAgent查询设备状态，确认CONVEYOR-01故障

3. 但用户要求"提交维修申请"——这是OA操作

4. 这里需要跨Agent协作：MESAgent把结果传给OAAgent，OAAgent调用 submit_maintenance 工具

5. 实际实现中可以用 LangGraph 编排多Agent工作流，本文为了简化用单Agent + 多工具实现

---

八、实施路线图

Phase	Task	Duration	Deliverable
Diagnosis	Map systems, identify integration points, select pilot	1-2 months	Architecture doc, API list
Pilot	Connect data, deploy 1 OA + 1 MES scenario	2-3 months	Working demo
Scale	Expand scenarios, improve RAG, establish operations	3-6 months	Full AI assistant
Optimize	Iterate models, expand new scenarios	Ongoing	Continuous improvement

---

九、踩坑经验

9.1 安全

• Least privilege: Agent accounts only get necessary permissions

• Audit logs: All AI operations logged (who, when, what)

• Double confirmation: Critical operations require human approval

9.2 性能

• Connection pooling: Reuse DB connections

• Caching: Cache frequent queries in Redis (TTL 30s)

• Async: Use queues for alarms, reports

9.3 模型 selection

Scenario	Model	Reason
Intent classification	GPT-4o-mini / local small model	Fast, cheap
Complex reasoning	GPT-4o / Claude-3.5	Strong, stable
On-premise	Qwen-72B / Ernie-4.0	Data stays local

9.4 Error handling

class SafeToolExecutor:
    def execute_with_fallback(self, tool_func, *args, **kwargs):
        try:
            return tool_func(*args, **kwargs)
        except ConnectionError:
            return "Connection timeout, please retry."
        except PermissionError:
            return "Permission denied, contact admin."
        except Exception as e:
            logger.error(f"Failed: {str(e)}")
            return f"Operation failed, logged for review."

---

十、总结

本文完整展示了AI Agent接入OA/MES的实操方案：

1. Layered architecture: Perception -> Reasoning -> Execution -> Feedback

2. Intent routing: Classifier dispatches to correct Agent

3. RAG enhancement: AI masters enterprise knowledge

4. Memory system: Milvus enables long-term coherent conversations

5. Tool orchestration: API integration with business systems

Next steps: Connect ERP/CRM/WMS, or add visual Agent for non-invasive operations.

---

References:

• LangChain Docs

• Milvus

• AI Agent Architecture

• Enterprise AI Integration

---

Like, bookmark, and follow for more practical content!