2026年,AI智能体(Agent)已从技术概念走到产业落地的关键节点。企业不再满足于"调用一下API"式的简单对话机器人,而是需要能理解业务流程、主动规划任务、调用外部工具、与其他Agent协作完成复杂目标的智能体系统。上海赢式信息科技有限公司(2010年成立,16年软件开发经验)自2023年起布局AI智能体开发,已为制造、物流、金融、教育等行业交付20+落地项目。本文系统讲解AI智能体从架构设计、Prompt工程、RAG知识库搭建、工具调用集成到部署上线的完整开发流程。

一、AI智能体技术架构:五大核心组件

一个完整可用的AI智能体系统由五大核心组件构成,理解它们的关系是架构设计的前提:

┌─────────────────────────────────────────────────────────────────────┐
│                        AI Agent 系统架构                              │
│                                                                     │
│  ┌─────────────────────────────────────────────────────────────┐   │
│  │  ⑤ 用户界面层 (Web/App/API/IM机器人)                         │   │
│  └─────────────────────────────────────────────────────────────┘   │
│                              ↓                                      │
│  ┌─────────────────────────────────────────────────────────────┐   │
│  │  ④ 编排调度层 (Orchestrator)                                 │   │
│  │    · 多Agent协作 · 任务分解 · 流程编排 · 状态管理              │   │
│  └─────────────────────────────────────────────────────────────┘   │
│                              ↓                                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐           │
│  │ ① LLM   │  │ ② 记忆   │  │ ③ 工具   │  │ ③ 知识   │           │
│  │  大脑   │  │  模块    │  │  调用    │  │  库(RAG) │           │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘           │
│       ↑             ↑             ↑              ↑                   │
│  ┌─────────────────────────────────────────────────────────────┐   │
│  │  基础设施: 向量数据库 / 工具注册表 / 消息队列 / 监控            │   │
│  └─────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘

1.1 LLM大脑

LLM是智能体的"大脑",负责理解用户意图、生成决策、推理规划。2026年企业级项目常用的LLM选型对比:

模型类型上下文窗口Function Calling成本 (每1M tokens)适用场景
GPT-4o闭源商用128K原生支持输入$5 / 输出$15通用高复杂场景
Claude 3.5 Sonnet闭源商用200K原生支持输入$3 / 输出$15长文档处理
通义千问Qwen2.5开源/商用128K支持约¥1/¥4 (商用API)中文场景 / 本地化部署
DeepSeek-V3开源/商用128K支持输入¥1/输出¥2代码生成 / 推理
Llama 3.1 70B开源可自部署128K需微调自部署硬件成本数据敏感 / 私有化

赢式科技的选型建议:数据不敏感场景优先商用API(稳定性和效果更好),数据敏感或有合规要求的场景走开源模型私有化部署。自部署的硬件门槛:7B模型需16GB显存,70B模型需4×80GB A100或等效GPU。

1.2 记忆模块

记忆模块让智能体"记住"过去的交互和信息。按作用域可分为三种:

  • 短期记忆(对话窗口):当前会话的上下文,直接通过发送历史消息给LLM实现。受上下文窗口大小限制(一般128K-200K tokens)。
  • 长期记忆(用户画像):跨会话持久化存储,如用户偏好、历史行为。一般用关系型数据库(PostgreSQL)或文档型数据库(MongoDB)存储。
  • 工作记忆(任务状态):当前正在执行的子任务状态、中间结果。用Redis缓存,任务完成后可清除。

二、Prompt工程实战

Prompt不是"随便写几句话",而是用结构化的方式引导LLM输出期望结果的工程化手段。以下是赢式科技团队总结的Prompt工程四大原则:

2.1 角色设定(Role Prompting)

system_prompt = """你是一个资深的工业MES系统工程师,拥有10年以上制造执行系统开发经验。
你的任务是帮助用户分析生产数据、排查工艺异常、优化产线节拍。
回答时必须遵循以下规则:
1. 先给出结论,再展开推理过程
2. 引用数据时注明具体来源和时间戳
3. 如果信息不足,主动追问需要补充哪些数据
4. 回答控制在500字以内,避免冗余表述
"""

2.2 Few-Shot示例

# Few-Shot让LLM通过示例理解输出格式和风格
examples = [
    {"input": "昨天A产线OEE是多少?", 
     "output": "A产线昨日OEE为72.3%,其中可用率89.1%、性能率91.4%、质量指数89.2%,较前日下降3.2个百分点,主要因23:15-00:05的设备故障停机30分钟。"},
    {"input": "帮我查一下3号工位最近的不良率",
     "output": "3号工位近7日不良率均值为2.18%,高于产线平均0.87%的1.31倍,建议重点关注。较高出现在8月6日(3.45%),最低在8月10日(1.12%)。"}
]

prompt = f"""你是MES数据分析师。请参考以下示例的风格回答问题:
示例1: Q={examples[0]['input']} A={examples[0]['output']}
示例2: Q={examples[1]['input']} A={examples[1]['output']}

用户问题: {user_query}
你的回答:"""

2.3 思维链(Chain-of-Thought)

# 让LLM先展示推理过程再给出最终答案
cot_prompt = """请一步步分析以下生产异常:
问题:今日09:30-10:00,B产线注塑机2号连续停机3次,每次5-8分钟。已采集数据:
- 停机时PLC报码 E-017(温度过高)
- 模具冷却水流量从12L/min降至3L/min
- 环境温度从28℃升至34℃

请按以下步骤分析:
步骤1:观察停机现象,列出可能的直接原因
步骤2:结合采集数据,排除不可能的原因
步骤3:得出最可能的根因
步骤4:给出排查建议

分析过程:
"""

测试数据显示,对需要多步推理的复杂问题,加入思维链提示可将准确率从约45%提升到78%以上。

2.4 结构化输出约束

# 用JSON Schema强制输出格式
from pydantic import BaseModel, Field
from typing import List, Optional

class ActionItem(BaseModel):
    tool_name: str = Field(description="要调用的工具名称")
    parameters: dict = Field(description="工具调用参数")
    reason: str = Field(description="调用该工具的原因说明")

class AgentResponse(BaseModel):
    analysis: str = Field(description="对用户问题的分析")
    actions: List[ActionItem] = Field(description="需要执行的工具调用列表")
    final_answer: Optional[str] = Field(description="如果可以直接回答则填,否则为null")

# 将JSON Schema作为System Prompt的一部分传给LLM
schema_prompt = f"""请严格按照以下JSON格式输出你的响应:
{AgentResponse.model_json_schema()}

不要输出任何JSON之外的文字,不要包含markdown代码块标记。"""

三、RAG知识库完整搭建流程

RAG(Retrieval-Augmented Generation,检索增强生成)是让智能体基于企业私有知识回答问题的核心技术。RAG的完整流水线分为5个阶段:

阶段1: 文档解析 → 阶段2: 文本切片 → 阶段3: 向量化 → 阶段4: 向量库存储 → 阶段5: 检索增强

┌──────────────┐   ┌──────────────┐   ┌──────────────┐   ┌──────────────┐   ┌──────────────┐
│ PDF/Word/HTML│→│ 语义切片策略  │→│ Embedding    │→│ 向量数据库   │→│ 相似度检索   │
│ Excel/图片    │   │ 按章节/段落   │   │ 模型编码     │   │ (Pinecone/   │   │ Top-K召回    │
│ 音频转写      │   │ 标题感知切片   │   │ (text-       │   │ Milvus/      │   │ + 重排模型    │
│               │   │ 300-500字/块  │   │ embedding-   │   │ Chroma)      │   │              │
│               │   │ 50-100字重叠  │   │ 3-small)     │   │              │   │              │
└──────────────┘   └──────────────┘   └──────────────┘   └──────────────┘   └──────────────┘

3.1 文档解析

企业知识源格式多样,赢式科技团队的解析方案:

# 文档解析常用工具对比
# PDF: PyMuPDF(fitz) → 速度最快, 但表格识别需要配合camelot-py
# Word: python-docx → 简单可靠
# Excel: openpyxl + 转为Markdown → 保留结构
# 图片: PaddleOCR → 中文OCR效果较优
# HTML: BeautifulSoup4 → 配合trafilatura提取正文

import fitz  # PyMuPDF
import re

def parse_pdf(file_path: str) -> list[dict]:
    doc = fitz.open(file_path)
    pages = []
    for page_num, page in enumerate(doc):
        text = page.get_text()
        # 保留标题层级标记
        text = re.sub(r'\n{3,}', '\n\n', text.strip())
        pages.append({
            "page": page_num + 1,
            "text": text,
            "source": file_path
        })
    doc.close()
    return pages

3.2 文本切片策略

切片质量直接影响检索效果。三种主流切片策略对比:

策略实现方式优点缺点推荐场景
固定长度切片每300字切一块,50字重叠实现简单可能切断语义快速原型
语义切片按段落/标题结构切保留语义完整性依赖文档格式规范企业内部文档
递归切片先按大结构切,超限再递归适应性较强实现复杂混合格式文档集
# LangChain的递归切片器实现(赢式科技实际项目中使用)
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    separators=["\n## ", "\n### ", "\n#### ", "\n\n", "\n", " ", ""],
    chunk_size=400,
    chunk_overlap=60,
    length_function=len,
    is_separator_regex=False,
)

chunks = splitter.split_documents(parsed_documents)
# 输出: List[Document], 每个Document包含page_content和metadata

3.3 向量化与向量库选型

# Embedding模型选型 (2026年)
# OpenAI: text-embedding-3-small (1536维, $0.02/1M tokens)
# 本地模型: BAAI/bge-m3 (1024维, 多语言, 可自部署)
# 中文优化: shibing624/text2vec-base-chinese (768维)

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-m3')
embeddings = model.encode(chunk_texts, batch_size=32, show_progress_bar=True)
# 输出: numpy array, shape=(num_chunks, 1024)
向量库类型单机QPS元数据过滤成本推荐场景
Pinecone全托管云服务10000+原生支持按存储+查询计费快速上线, 无需运维
Milvus开源分布式5000+原生支持服务器成本大规模私有化部署
ChromaDB开源嵌入式500支持免费原型验证/中小项目
pgvectorPostgreSQL插件2000原生支持现有PG服务器已有PG基础设施
FAISSMeta开源库10000+需自行实现免费纯向量检索, 不关心元数据

3.4 检索与生成

# 完整的RAG检索增强生成流程
def rag_query(user_question: str, top_k: int = 5) -> str:
    # 步骤1: 问题向量化
    question_embedding = model.encode([user_question])[0]

    # 步骤2: 向量库检索 (HNSW算法, 近似最近邻)
    search_results = vector_db.search(
        vector=question_embedding,
        top_k=top_k,
        filter={"source_type": "manual"},  # 可选元数据过滤
        metric="cosine"
    )

    # 步骤3: 重排模型精排 (可选但效果显著)
    reranked = rerank_model.rerank(
        query=user_question,
        documents=[r["text"] for r in search_results]
    )[:3]

    # 步骤4: 构建增强Prompt
    context = "\n\n---\n\n".join([r["text"] for r in reranked])
    enhanced_prompt = f"""基于以下参考资料回答用户问题。
如果参考资料中没有相关信息,请说"根据现有知识库无法回答此问题",
不要编造答案。

参考资料:
{context}

用户问题: {user_question}

请给出准确、简洁的回答:"""

    # 步骤5: 调用LLM生成回答
    response = llm_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个专业的企业知识问答助手。"},
            {"role": "user", "content": enhanced_prompt}
        ],
        temperature=0.2,  # RAG场景温度要低
        max_tokens=1024
    )

    return response.choices[0].message.content

四、工具调用 Function Calling 实践

Function Calling让LLM能决定"什么时候调用什么工具、传什么参数",是智能体具备行动能力的关键。赢式科技团队的实现方案:

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_mes_production_data",
            "description": "查询MES系统的生产数据,包括产量、OEE、不良率、停机记录等",
            "parameters": {
                "type": "object",
                "properties": {
                    "metric": {
                        "type": "string",
                        "enum": ["production", "oee", "defect_rate", "downtime", "yield"],
                        "description": "要查询的指标类型"
                    },
                    "time_range": {
                        "type": "object",
                        "properties": {
                            "start": {"type": "string", "description": "开始时间, 格式YYYY-MM-DD"},
                            "end": {"type": "string", "description": "结束时间, 格式YYYY-MM-DD"}
                        },
                        "required": ["start", "end"]
                    },
                    "line_id": {
                        "type": "string",
                        "description": "产线ID, 如 'A', 'B', 'C'. 不填则查询全部产线"
                    }
                },
                "required": ["metric", "time_range"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_work_order",
            "description": "向ERP系统发送工单",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_code": {"type": "string", "description": "产品编码"},
                    "quantity": {"type": "integer", "description": "生产数量"},
                    "priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]},
                    "line_id": {"type": "string"}
                },
                "required": ["product_code", "quantity", "priority"]
            }
        }
    }
]

# 调用示例: LLM自动决定是否调用工具
response = llm_client.chat.completions.create(
    model="gpt-4o",
    messages=conversation_history,
    tools=tools,
    tool_choice="auto"  # 让LLM自动决定
)

# 处理工具调用
message = response.choices[0].message
if message.tool_calls:
    for tool_call in message.tool_calls:
        if tool_call.function.name == "query_mes_production_data":
            args = json.loads(tool_call.function.arguments)
            result = mes_client.query(args)
            conversation_history.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

五、多Agent协作模式

复杂业务场景往往需要多个Agent分工协作。三种主流协作模式及适用场景:

模式架构描述优点缺点适用场景
编排者-执行者 (Orchestrator-Worker)中央Agent拆解任务分派给专业Agent流程可控, 结果可靠编排者成为瓶颈任务可明确分解的场景
对等协作 (Peer-to-Peer)多个Agent直接沟通协作灵活, 适应性强对话冗长, 成本高需求模糊, 需多方协商
层级团队 (Hierarchical Team)管理者Agent + 工作组Agent + 专家Agent复杂项目可控实现复杂大型项目, 需严格流程
# 赢式科技实际项目:MES异常根因分析多Agent系统
# 编排者: 根因分析Agent (协调各专家Agent)
# 工作者1: PLC数据采集Agent (调用PLC数据API)
# 工作者2: 工艺参数分析Agent (分析温度/压力/节拍)
# 工作者3: 设备维护Agent (查询设备历史维护记录)
# 工作者4: 知识库Agent (RAG检索维修手册)

class OrchestratorAgent:
    def __init__(self):
        self.plc_agent = PlcDataAgent()
        self.process_agent = ProcessAnalysisAgent()
        self.maintenance_agent = MaintenanceAgent()
        self.rag_agent = KnowledgeBaseAgent()
        self.llm = LLMClient("gpt-4o")

    async def analyze(self, anomaly: str) -> dict:
        # 步骤1: LLM根据异常描述决定需要哪些专家介入
        plan = await self.llm.plan(
            task=f"分析异常: {anomaly}",
            available_agents=["plc", "process", "maintenance", "rag"]
        )
        # 步骤2: 并行调用被选中的专家Agent
        results = await asyncio.gather(*[
            self._dispatch(agent_type, anomaly) for agent_type in plan.needed_agents
        ])
        # 步骤3: LLM综合各Agent结果得出根因
        final_analysis = await self.llm.synthesize(results, anomaly)
        return final_analysis

六、部署方案与性能对比

AI智能体的部署方式取决于数据敏感度、延迟要求和预算约束。三种主流方案实测数据:

部署方式数据离开内网首token延迟并发能力月成本 (估算)典型方案
纯云API300-800ms由服务商决定¥500-5000 (按量)OpenAI API / 通义千问API
混合云 (Embedding本地化)部分400-900ms云侧决定 + 本地Embedding无压力¥3000-10000pgvector本地 + GPT-4o API
全本地化1500-3000ms (70B模型)取决于GPU数量服务器一次性投入4×A100 + vLLM推理

6.1 生产环境部署架构

# 赢式科技生产级AI智能体部署架构
┌─────────────────────────────────────────────────────────┐
│                    负载均衡层 (Nginx)                      │
│  · SSL终止 · 限流 · WAF防护                               │
└─────────────────────────────────────────────────────────┘
                         ↓
┌─────────────────────────────────────────────────────────┐
│                 API网关 (Kong / APISIX)                   │
│  · 认证鉴权 · 路由转发 · 监控埋点 · 熔断降级               │
└─────────────────────────────────────────────────────────┘
                         ↓
┌─────────────────────────────────────────────────────────┐
│              Agent服务层 (Python FastAPI)                 │
│  · Agent Orchestrator · Function Router · RAG Pipeline    │
│  · 多实例水平扩展 (K8s Deployment, replicas=3+)           │
└─────────────────────────────────────────────────────────┘
         ↕                ↕                ↕
┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐
│ LLM推理服务   │  │ 向量数据库    │  │ 企业内部系统 (MES/   │
│ (vLLM/TGI)   │  │ (Milvus)     │  │ ERP/PLC/数据库)      │
└──────────────┘  └──────────────┘  └──────────────────────┘
         ↕                ↕
┌──────────────┐  ┌──────────────┐
│ Redis缓存    │  │ 监控日志系统  │
│ (会话/短期记忆)│  │ (Prometheus/ │
│              │  │  ELK/LangSmith)│
└──────────────┘  └──────────────┘

6.2 监控指标

生产环境AI智能体需要监控的关键指标:

指标类别具体指标告警阈值建议
性能首token延迟 (TTFT)> 2s 告警
性能总响应延迟> 5s 告警
性能并发请求数> 单实例QPS×0.8 告警
质量用户满意度评分< 3.5/5 触发人工review
质量幻觉率 (需人工标注抽样)> 5% 触发Prompt优化
成本日均Token消耗超过预算120% 告警
系统Function Calling成功率< 90% 告警

七、赢式科技AI智能体低代码平台

上海赢式信息科技有限公司(2010年成立,16年软件开发经验)在2023年启动AI智能体低代码平台研发,2025年正式对外提供平台化服务。平台核心能力包括:

  • 可视化Agent编排界面:拖拽式流程设计器,支持条件分支、并行执行、循环等控制流,非技术人员也能搭建业务Agent
  • 预置行业知识库模板:工业制造(MES/PLC/工艺)、仓储物流(WMS/TMS)、金融客服、教育培训四大行业已提供开箱即用的知识库模板
  • 通用工具市场:预置50+常用工具连接器,包括数据库查询、API调用、邮件发送、Excel处理、Web搜索等
  • 模型多租户管理:支持租户级别的模型配置(选GPT-4o还是通义千问、是否启用Function Calling、温度参数等)
  • 全链路监控与回放:每次Agent执行的完整trace可回放查看,支持LLM调用成本分析

目前赢式科技已为20+企业交付AI智能体项目,典型客户包括某头部汽车零部件企业(MES数据分析Agent,月处理工单5000+)、某第三方物流企业(智能调度Agent,提升调度效率40%)、某制造业上市公司(设备预测性维护Agent,降低非计划停机35%)。