🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度

最近在推进公司一个核心业务系统的智能化改造，团队内部讨论最多的就是： “我们这套复杂的微服务架构，到底该怎么接AI？” 直接调用大模型API，回答质量不稳定，还容易“一本正经地胡说八道”；想让它操作内部系统，又担心安全和权限问题。经过几个月的摸索和落地，我们总结出一套结合 Agent（智能体）、RAG（检索增强生成）和 MCP（模型上下文协议） 的企业级改造方案。今天，我就把这套方案的完整思路、技术选型、实战代码和踩过的坑，毫无保留地分享给大家。

无论你是想为现有系统增加AI能力的技术负责人，还是正在探索AI落地的开发者，这篇文章都能给你提供一个从架构设计到代码实现的完整闭环参考。我们会从零开始，构建一个能理解业务、操作数据库、并安全调用内部API的智能体。

1. 背景与核心概念：为什么需要 Agent + RAG + MCP？

在深入技术细节之前，我们必须先理清这三个核心概念，以及它们组合起来能解决企业级项目的哪些痛点。

1.1 企业级AI集成的三大挑战

1. 知识时效性与准确性 ：大模型的训练数据有截止日期，无法知晓你公司最新的产品文档、技术规范或客户数据。直接提问，它可能给出过时或错误的答案。
2. 操作与执行能力 ：大模型本质是“思考者”，不是“执行者”。它无法直接查询你的数据库、调用内部审批接口或重启服务器。
3. 安全与权限管控 ：企业系统涉及敏感数据和核心业务，不可能将内部API或数据库直接暴露给外部AI服务。

1.2 核心概念拆解

Agent（智能体） 你可以把它理解为一个“AI项目经理”。它接收用户的任务（如“查询上个月A产品的销售额”），然后进行规划、思考、决策，并调用各种工具（Tools）来完成任务。Agent的核心能力是 自主规划与工具调用 。

RAG（检索增强生成） 这是解决大模型“知识陈旧”和“幻觉”问题的关键技术。其核心流程是：

1. 检索（Retrieval） ：当用户提问时，先从你的专属知识库（如产品文档、公司制度PDF、数据库）中搜索相关片段。
2. 增强（Augmentation） ：将搜索到的相关文本作为“参考材料”，和用户问题一起提交给大模型。
3. 生成（Generation） ：大模型基于“参考材料”生成更准确、更相关的回答。

RAG让大模型的回答 有据可依 ，极大地提升了专业领域的回答质量。

MCP（模型上下文协议，Model Context Protocol） 这是一个由Anthropic等公司推动的 新兴开放协议 。它旨在标准化大模型与外部数据源、工具之间的连接方式。你可以把它想象成AI世界的“USB标准协议”。

• 对AI模型 ：MCP提供了一个统一的方式来发现、描述和调用外部工具（如数据库、API、文件系统）。
• 对开发者 ：你可以编写一个MCP Server，将你的内部系统（如CRM、ERP）的能力“包装”成标准化的工具，然后安全地提供给任何支持MCP协议的AI Agent使用。

1.3 三者如何协同工作？

想象一个场景：业务人员问AI：“帮我分析一下客户‘张三’最近的订单情况，如果有异常订单，就通知他的客户经理。”

1. Agent 作为大脑，解析这个复杂任务，将其拆解为子任务：a) 查询客户信息，b) 查询订单，c) 分析异常，d) 发送通知。
2. RAG 在步骤a和b中发挥作用。当Agent需要查询时，它可以通过RAG从企业知识库（如客户数据库视图的说明文档）中检索“如何正确查询客户订单”的SQL示例或API规范，确保操作准确。
3. MCP 在步骤a, b, d中提供“手和脚”。Agent通过MCP协议，调用已对接好的“客户查询工具”、“订单查询工具”和“内部消息通知工具”来实际执行操作。所有操作都在受控的安全边界内进行。

简单总结 ： Agent负责思考和规划，RAG负责提供精准的知识参考，MCP负责提供安全、标准的执行通道。 三者结合，构成了一个既能“懂行”又能“办事”的企业级AI应用骨架。

2. 环境准备与项目结构

我们将以一个简化的“电商订单分析”场景为例，演示如何构建这套系统。技术栈选择目前生态最活跃的Python体系。

2.1 基础环境

• 操作系统 ：macOS / Linux (推荐) 或 WSL2 (Windows)
• Python版本 ：>= 3.10
• 包管理工具 ： pip 或 poetry (本文使用 pip )
• IDE ：VS Code 或 PyCharm，安装Python插件。

2.2 核心库与版本说明

我们将使用以下主流开源库，它们的版本迭代较快，以下版本在撰写时已验证可用，请根据实际情况调整。

# 创建项目目录并进入
mkdir enterprise-ai-agent && cd enterprise-ai-agent
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装核心依赖
pip install "langchain>=0.1.0" "langchain-community>=0.0.10"  # Agent和RAG框架
pip install "langchain-openai>=0.0.5"  # OpenAI模型集成
pip install chromadb>=0.4.22  # 向量数据库，用于存储和检索知识
pip install tiktoken  # Token计数
pip install "pydantic>=2.0.0"  # 数据验证
pip install python-dotenv  # 管理环境变量
pip install fastapi>=0.104.0 uvicorn>=0.24.0  # 用于构建MCP Server (HTTP Transport)
pip install mcp>=0.1.0  # MCP协议Python SDK (如果可用，否则用标准方式)
# 注意：正式的 `mcp` Python包可能还在早期，我们后续会演示其概念实现。

2.3 项目结构

我们的项目将按功能模块进行组织，清晰易懂。

enterprise-ai-agent/
├── .env                    # 存储API密钥等敏感配置
├── requirements.txt        # 项目依赖
├── app.py                  # 主应用入口，Agent调度中心
├── config.py               # 配置文件
├── knowledge_base/         # RAG知识库相关
│   ├── docs/              # 存放原始知识文档（.txt, .md, .pdf）
│   ├── loaders.py         # 文档加载器
│   ├── vector_store.py    # 向量数据库初始化与检索
│   └── retriever.py       # 检索器封装
├── agents/                 # Agent定义
│   ├── order_analyst.py   # 订单分析智能体
│   └── tools.py           # Agent可用的工具定义
├── mcp_servers/           # MCP服务器（模拟内部工具）
│   ├── order_server.py    # 订单查询MCP服务
│   └── notification_server.py # 通知服务MCP服务
└── utils/
    └── helpers.py         # 通用工具函数

3. 第一步：构建企业知识库（RAG实战）

首先，我们解决“知识”问题，为AI准备好准确的参考资料。

3.1 准备知识文档

在 knowledge_base/docs/ 下创建几个示例文档。

knowledge_base/docs/product_policy_2024.md

# 2024年产品异常订单判定政策

## 异常订单定义
1.  **价格异常**：订单金额低于产品历史最低价（LTV）的70%。
2.  **数量异常**：单次购买数量超过该客户过去30天平均购买量的5倍。
3.  **地址异常**：收货地址与客户常用地址（过去90%的订单）不一致，且未在备注中说明。
4.  **支付异常**：使用非绑定支付方式，且金额大于5000元。

## 处理流程
发现异常订单后，系统应：
1.  自动标记订单状态为“待审核”。
2.  向订单所属的客户经理发送站内通知。
3.  通知内容模板：[异常订单警报] 客户{客户名}，订单号{订单号}，异常类型：{类型}，请及时处理。

knowledge_base/docs/customer_db_schema.txt

表名: customers
- id: INT, 主键，客户ID
- name: VARCHAR(100), 客户名称
- customer_manager_id: INT, 客户经理ID
- common_shipping_address: TEXT, 常用收货地址
- vip_level: INT, 会员等级 (1-5)

表名: orders
- order_id: VARCHAR(50), 主键，订单号
- customer_id: INT, 外键，关联customers.id
- product_id: INT, 产品ID
- quantity: INT, 购买数量
- amount: DECIMAL(10,2), 订单金额
- shipping_address: TEXT, 本次收货地址
- status: VARCHAR(20), 状态 ('pending', 'shipped', 'cancelled', 'under_review')
- created_at: DATETIME, 创建时间

3.2 实现文档加载与向量化

我们使用 ChromaDB 作为向量数据库， OpenAI 的嵌入模型来将文本转换为向量。

knowledge_base/loaders.py

import os
from langchain_community.document_loaders import TextLoader, UnstructuredMarkdownLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from config import settings

def load_and_split_documents(docs_dir: str):
    """加载指定目录下的所有文档并进行文本分割"""
    documents = []
    for filename in os.listdir(docs_dir):
        file_path = os.path.join(docs_dir, filename)
        if filename.endswith('.txt'):
            loader = TextLoader(file_path, encoding='utf-8')
        elif filename.endswith('.md'):
            loader = UnstructuredMarkdownLoader(file_path)
        else:
            continue  # 可扩展支持PDF、Word等
        loaded_docs = loader.load()
        documents.extend(loaded_docs)
    
    # 文本分割：避免文档过长，超出模型上下文
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,  # 每个片段约1000字符
        chunk_overlap=200, # 片段间重叠200字符，保持语义连贯
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
    )
    split_docs = text_splitter.split_documents(documents)
    print(f"共加载 {len(documents)} 个文档，分割为 {len(split_docs)} 个片段。")
    return split_docs

def create_vector_store(documents, persist_directory="./chroma_db"):
    """创建并持久化向量存储"""
    # 初始化嵌入模型
    embeddings = OpenAIEmbeddings(
        model="text-embedding-3-small",  # 或 "text-embedding-ada-002"
        openai_api_key=settings.OPENAI_API_KEY
    )
    
    # 创建向量数据库
    vector_store = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory=persist_directory
    )
    vector_store.persist()  # 持久化到磁盘
    print(f"向量数据库已创建并保存至 {persist_directory}")
    return vector_store

if __name__ == "__main__":
    # 测试代码
    from dotenv import load_dotenv
    load_dotenv()
    docs = load_and_split_documents("./docs")
    vs = create_vector_store(docs)

config.py

import os
from dotenv import load_dotenv

load_dotenv()  # 从 .env 文件加载环境变量

class Settings:
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    # 其他配置如数据库连接等可以在这里添加
    
settings = Settings()

.env 文件

OPENAI_API_KEY=你的OpenAI_API密钥

3.3 构建检索器

knowledge_base/retriever.py

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from config import settings

class KnowledgeRetriever:
    def __init__(self, persist_directory="./chroma_db"):
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            openai_api_key=settings.OPENAI_API_KEY
        )
        self.vector_store = Chroma(
            persist_directory=persist_directory,
            embedding_function=self.embeddings
        )
        # 创建检索器，可以设置相似度阈值或返回数量
        self.retriever = self.vector_store.as_retriever(
            search_kwargs={"k": 4}  # 返回最相关的4个片段
        )
    
    def query(self, question: str):
        """检索与问题相关的知识片段"""
        relevant_docs = self.retriever.get_relevant_documents(question)
        return relevant_docs
    
    def get_context_for_question(self, question: str) -> str:
        """将检索到的知识片段组合成上下文文本"""
        docs = self.query(question)
        context = "\n\n---\n\n".join([doc.page_content for doc in docs])
        return context

# 全局检索器实例
retriever = KnowledgeRetriever()

现在，运行 loaders.py 的 if __name__ == "__main__" 部分，你的知识文档就会被处理并存入本地的 chroma_db 目录。后续可以直接使用 retriever 来查询相关知识。

4. 第二步：定义与封装内部工具（MCP思想实践）

MCP的核心是“工具标准化”。我们虽然不直接使用官方的MCP SDK（因其仍在演进），但完全遵循其设计思想：将内部能力封装成具有清晰输入输出描述的、可被AI发现和调用的工具。

4.1 模拟内部工具：订单查询与通知

我们先模拟两个内部服务，它们通常以REST API或gRPC形式存在。

mcp_servers/order_server.py (模拟订单查询服务)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(title="Internal Order Service")

# 模拟内存数据库
mock_customers = [
    {"id": 1, "name": "张三", "customer_manager_id": 101, "vip_level": 3},
    {"id": 2, "name": "李四", "customer_manager_id": 102, "vip_level": 1},
]
mock_orders = [
    {"order_id": "ORD001", "customer_id": 1, "amount": 1500.00, "quantity": 2, "status": "shipped", "shipping_address": "北京市海淀区"},
    {"order_id": "ORD002", "customer_id": 1, "amount": 99.00, "quantity": 1, "status": "under_review", "shipping_address": "上海市浦东新区"}, # 地址异常
    {"order_id": "ORD003", "customer_id": 2, "amount": 5000.00, "quantity": 10, "status": "pending", "shipping_address": "广州市天河区"},
]

class OrderQuery(BaseModel):
    customer_name: Optional[str] = None
    order_id: Optional[str] = None
    min_amount: Optional[float] = None

@app.post("/api/internal/orders/query")
async def query_orders(params: OrderQuery):
    """内部订单查询接口"""
    filtered_orders = mock_orders
    if params.customer_name:
        # 根据客户名找到ID
        customer_ids = [c["id"] for c in mock_customers if params.customer_name in c["name"]]
        filtered_orders = [o for o in filtered_orders if o["customer_id"] in customer_ids]
    if params.order_id:
        filtered_orders = [o for o in filtered_orders if params.order_id == o["order_id"]]
    if params.min_amount:
        filtered_orders = [o for o in filtered_orders if o["amount"] >= params.min_amount]
    
    return {"code": 0, "msg": "success", "data": filtered_orders}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)

mcp_servers/notification_server.py (模拟内部通知服务)

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

app = FastAPI(title="Internal Notification Service")

class NotificationRequest(BaseModel):
    manager_id: int
    title: str
    content: str
    priority: str = "normal"  # low, normal, high

@app.post("/api/internal/notification/send")
async def send_notification(req: NotificationRequest):
    """内部通知发送接口"""
    # 这里模拟发送通知，实际可能调用消息队列、邮件或IM服务
    print(f"[模拟通知发送] 给经理 {req.manager_id}: [{req.priority}] {req.title} - {req.content}")
    return {"code": 0, "msg": "通知已发送", "data": {"notification_id": "mock_123"}}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8002)

分别运行这两个文件，你就启动了两个模拟的内部服务。

4.2 为Agent封装标准化工具

现在，我们将这两个HTTP服务封装成LangChain Agent能识别的 Tool 对象。关键在于提供清晰、结构化的工具描述，让AI能理解何时以及如何使用它。

agents/tools.py

import requests
from typing import Optional, Dict, Any
from langchain.tools import BaseTool, Tool
from pydantic import BaseModel, Field
from config import settings

# --- 1. 订单查询工具 ---
class OrderQueryInput(BaseModel):
    """订单查询工具的输入参数"""
    customer_name: Optional[str] = Field(None, description="客户姓名，用于筛选订单")
    order_id: Optional[str] = Field(None, description="订单号，精确查询")
    min_amount: Optional[float] = Field(None, description="最低订单金额，用于筛选")

class OrderQueryTool(BaseTool):
    name = "query_orders"
    description = """
    用于查询公司内部的订单数据。
    当用户需要了解客户订单情况、查找特定订单或分析订单数据时使用此工具。
    输入可以是客户名、订单号或金额范围。
    """
    args_schema = OrderQueryInput
    
    def _run(self, customer_name: Optional[str] = None, order_id: Optional[str] = None, min_amount: Optional[float] = None) -> str:
        """执行工具调用"""
        try:
            payload = {}
            if customer_name:
                payload["customer_name"] = customer_name
            if order_id:
                payload["order_id"] = order_id
            if min_amount:
                payload["min_amount"] = min_amount
            
            # 调用模拟的内部订单服务
            resp = requests.post(
                "http://localhost:8001/api/internal/orders/query",
                json=payload,
                timeout=10
            )
            resp.raise_for_status()
            result = resp.json()
            if result["code"] == 0:
                orders = result["data"]
                if not orders:
                    return "未找到符合条件的订单。"
                # 将结果格式化为易读的字符串
                formatted = []
                for o in orders:
                    formatted.append(f"订单号: {o['order_id']}, 金额: {o['amount']}, 状态: {o['status']}, 收货地址: {o.get('shipping_address', 'N/A')}")
                return "查询到的订单：\n" + "\n".join(formatted)
            else:
                return f"查询失败：{result.get('msg')}"
        except requests.exceptions.RequestException as e:
            return f"网络请求失败：{str(e)}"
        except Exception as e:
            return f"工具执行异常：{str(e)}"

# --- 2. 通知发送工具 ---
class NotificationInput(BaseModel):
    """通知发送工具的输入参数"""
    manager_id: int = Field(..., description="接收通知的客户经理ID")
    title: str = Field(..., description="通知标题")
    content: str = Field(..., description="通知正文内容")
    priority: str = Field("normal", description="通知优先级：low, normal, high")

class NotificationTool(BaseTool):
    name = "send_notification"
    description = """
    向指定的客户经理发送内部通知。
    当需要提醒、告警或通知客户经理处理某项事务时使用此工具。
    必须提供经理ID、标题和内容。
    """
    args_schema = NotificationInput
    
    def _run(self, manager_id: int, title: str, content: str, priority: str = "normal") -> str:
        try:
            payload = {
                "manager_id": manager_id,
                "title": title,
                "content": content,
                "priority": priority
            }
            resp = requests.post(
                "http://localhost:8002/api/internal/notification/send",
                json=payload,
                timeout=10
            )
            resp.raise_for_status()
            result = resp.json()
            return f"通知发送结果：{result['msg']}"
        except requests.exceptions.RequestException as e:
            return f"通知发送失败：{str(e)}"
        except Exception as e:
            return f"工具执行异常：{str(e)}"

# --- 3. 知识检索工具 (RAG集成) ---
from knowledge_base.retriever import retriever

class KnowledgeQueryInput(BaseModel):
    question: str = Field(..., description="需要查询的公司政策、流程或数据规范问题")

class KnowledgeQueryTool(BaseTool):
    name = "query_knowledge_base"
    description = """
    查询公司内部的知识库，包括政策文档、流程规范、数据库Schema等。
    当需要确认公司规定、操作流程或数据定义时，优先使用此工具获取准确信息。
    """
    args_schema = KnowledgeQueryInput
    
    def _run(self, question: str) -> str:
        """从向量知识库中检索相关信息"""
        try:
            context = retriever.get_context_for_question(question)
            if not context or len(context.strip()) < 10:
                return "知识库中未找到相关信息。"
            # 注意：这里只是返回检索到的原始知识片段。
            # 在实际的Agent流程中，这个上下文会和大模型结合，生成最终答案。
            return f"从知识库中检索到以下相关信息：\n\n{context}"
        except Exception as e:
            return f"知识检索失败：{str(e)}"

# 导出工具列表
def get_all_tools():
    """获取所有已定义的工具，供Agent使用"""
    return [
        OrderQueryTool(),
        NotificationTool(),
        KnowledgeQueryTool(),
    ]

至此，我们已经完成了 RAG知识库 的构建和 MCP思想下的工具封装 。接下来，我们将让Agent作为大脑，来协调使用这些能力。

5. 第三步：构建智能体（Agent）与任务编排

我们将使用LangChain的ReAct框架来构建一个能自主规划、使用工具的智能体。

agents/order_analyst.py

import os
from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from agents.tools import get_all_tools
from config import settings

class OrderAnalystAgent:
    def __init__(self):
        # 1. 初始化大语言模型
        self.llm = ChatOpenAI(
            model="gpt-4o",  # 或 "gpt-3.5-turbo"，gpt-4o在工具调用上更准确
            temperature=0,    # 温度设为0，使输出更确定、更可靠
            openai_api_key=settings.OPENAI_API_KEY
        )
        
        # 2. 获取所有可用工具
        self.tools = get_all_tools()
        
        # 3. 定义ReAct代理的提示词模板
        # 这个模板至关重要，它指导Agent如何思考、何时使用工具、如何总结。
        self.prompt_template = PromptTemplate.from_template("""
你是一个专业的电商订单分析AI助手，负责帮助业务人员分析订单数据、识别异常并触发相应流程。

你拥有以下工具：
{tools}

在回答用户问题时，请严格遵循以下步骤：
1.  **理解与规划**：首先，理解用户的真实意图。如果问题涉及公司内部政策、流程或数据定义（例如“什么是异常订单？”），你必须先使用 `query_knowledge_base` 工具查询知识库，确保你的操作符合公司规定。
2.  **执行与检索**：根据规划，使用相应的工具。如果需要查询订单数据，使用 `query_orders`。如果需要发送通知，使用 `send_notification`。每次使用工具时，必须提供工具要求的所有参数。
3.  **分析与总结**：根据工具返回的结果，结合你的分析，给出最终的回答。如果发现异常订单（参考知识库中的定义），你应该在回答中明确指出，并建议或直接使用 `send_notification` 工具通知客户经理。

请特别注意：
- 所有操作必须基于事实和工具返回的数据，不要编造信息。
- 使用工具时，输入参数必须准确。例如，`send_notification` 需要明确的 `manager_id`。
- 你的最终输出应该是清晰、有条理的中文回答。

现在，开始处理以下问题：
问题：{input}

请按步骤思考，你可以使用工具。你的思考过程应放在 `Thought:`、`Action:`、`Action Input:`、`Observation:` 标签内。
最终答案应放在 `Final Answer:` 标签内。
{agent_scratchpad}  # LangChain会自动填充工具执行的中间记录
""")
        
        # 4. 创建ReAct Agent
        self.agent = create_react_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=self.prompt_template
        )
        
        # 5. 创建Agent执行器，控制交互流程
        self.agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            verbose=True,  # 设为True可以看到Agent的详细思考过程，生产环境可设为False
            handle_parsing_errors=True,  # 优雅处理解析错误
            max_iterations=5,  # 防止无限循环
            early_stopping_method="generate"  # 当Agent认为可以给出最终答案时停止
        )
    
    def run(self, query: str) -> str:
        """执行用户查询"""
        try:
            result = self.agent_executor.invoke({"input": query})
            return result["output"]
        except Exception as e:
            return f"Agent执行过程中出现错误：{str(e)}"

# 创建全局Agent实例
analyst_agent = OrderAnalystAgent()

6. 第四步：主程序与完整流程演示

现在，我们把所有部分串联起来，形成一个完整的应用。

app.py

import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from agents.order_analyst import analyst_agent
import subprocess
import time
from threading import Thread

def start_mcp_servers():
    """在后台启动模拟的MCP服务器（内部工具服务）"""
    def run_server(script_name, port):
        # 这里简化处理，实际生产环境会用进程管理工具如supervisor
        print(f"启动 {script_name} 在端口 {port}...")
        # 注意：这是演示，实际应使用稳定的后台进程管理
        # subprocess.Popen([sys.executable, f"mcp_servers/{script_name}.py"])
        pass
    
    # 在实际演示中，我们假设服务已经手动启动（见第4.1节）
    # run_server("order_server", 8001)
    # run_server("notification_server", 8002)
    print("提示：请确保已分别在两个终端中运行了 order_server.py 和 notification_server.py")
    time.sleep(1)

def main():
    print("=" * 50)
    print("企业级AI订单分析助手启动")
    print("=" * 50)
    
    # 1. 启动后台服务（模拟）
    start_mcp_servers()
    
    # 2. 初始化Agent
    print("\n初始化智能体...")
    # analyst_agent 已在导入时初始化
    
    # 3. 交互循环
    print("\n你可以输入以下示例问题，或输入 'quit' 退出：")
    print("1. 查询客户‘张三’的所有订单")
    print("2. 帮我分析一下张三最近的订单有没有异常？")
    print("3. 如果有异常，通知他的客户经理")
    
    while True:
        try:
            user_input = input("\n>>> 请输入你的问题: ").strip()
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("再见！")
                break
            if not user_input:
                continue
            
            print("\n" + "="*30 + " 思考过程 " + "="*30)
            # 执行查询
            answer = analyst_agent.run(user_input)
            print("\n" + "="*30 + " 最终回答 " + "="*30)
            print(answer)
            
        except KeyboardInterrupt:
            print("\n程序被中断。")
            break
        except Exception as e:
            print(f"发生未知错误：{e}")

if __name__ == "__main__":
    # 确保环境变量已加载
    from dotenv import load_dotenv
    load_dotenv()
    
    if not os.getenv("OPENAI_API_KEY"):
        print("错误：请在项目根目录的 .env 文件中设置 OPENAI_API_KEY")
        sys.exit(1)
        
    main()

6.1 完整运行演示

1. 准备环境 ：确保 .env 文件中的 OPENAI_API_KEY 已设置。
2. 启动知识库 ：运行一次 knowledge_base/loaders.py 以构建向量数据库。
3. 启动内部服务 ：打开两个终端，分别运行：

   # 终端1
   python mcp_servers/order_server.py
   # 终端2
   python mcp_servers/notification_server.py
   

4. 运行主程序 ：

   python app.py
   

6.2 示例交互

输入 ： 帮我分析一下张三最近的订单有没有异常？

Agent的思考过程（verbose=True时可见） ：

Thought: 用户想分析客户“张三”的订单是否有异常。首先，我需要明确“异常订单”的定义，这属于公司政策，我应该先查询知识库。
Action: query_knowledge_base
Action Input: {"question": "异常订单的定义是什么？判定标准有哪些？"}
Observation: 从知识库中检索到以下相关信息：
# 2024年产品异常订单判定政策
## 异常订单定义
1.  **价格异常**：订单金额低于产品历史最低价（LTV）的70%...
2.  **数量异常**：单次购买数量超过该客户过去30天平均购买量的5倍...
3.  **地址异常**：收货地址与客户常用地址（过去90%的订单）不一致...
4.  **支付异常**：使用非绑定支付方式，且金额大于5000元...
---
表名: customers
- id: INT, 主键，客户ID
- name: VARCHAR(100), 客户名称...
表名: orders...

Thought: 现在我了解了异常的定义。接下来，我需要获取张三的订单数据来进行分析。
Action: query_orders
Action Input: {"customer_name": "张三"}
Observation: 查询到的订单：
订单号: ORD001, 金额: 1500.0, 状态: shipped, 收货地址: 北京市海淀区
订单号: ORD002, 金额: 99.0, 状态: under_review, 收货地址: 上海市浦东新区

Thought: 我查到了张三的两条订单。ORD002的状态已经是“under_review”（待审核），这很可能就是被标记为异常的订单。根据知识库，地址异常是指“收货地址与客户常用地址不一致”。ORD001的地址是北京，ORD002是上海。假设北京是常用地址，那么ORD002就符合地址异常。另外，金额99元，可能需要检查是否属于价格异常。我需要通知客户经理。但首先，我需要知道张三的客户经理ID。
（注：这里Agent可能会尝试从知识库或工具中获取manager_id，我们的模拟数据中，张三的manager_id是101）
Action: send_notification
Action Input: {"manager_id": 101, "title": "异常订单警报", "content": "客户张三，订单号ORD002，异常类型：地址异常（收货地址上海与常用地址北京不符），请及时处理。", "priority": "normal"}
Observation: 通知发送结果：通知已发送

Final Answer: 已完成对客户“张三”订单的分析。
1.  **查询结果**：共找到2条订单。
    - ORD001: 金额1500元，状态“已发货”，收货地址北京。
    - ORD002: 金额99元，状态“待审核”，收货地址上海。
2.  **异常分析**：
    - 订单ORD002被系统标记为“待审核”，疑似异常。
    - 根据公司《异常订单判定政策》，该订单的**收货地址（上海）** 与客户张三的常用地址（假设为北京）不一致，触发“地址异常”规则。
    - 订单金额99元，需结合产品历史最低价进一步判断是否属于“价格异常”。
3.  **已执行操作**：已根据流程，使用内部通知系统向客户经理（ID: 101）发送了异常订单警报。
4.  **建议**：请客户经理登录系统查看订单ORD002的详细信息，并进行后续处理。

通过这个完整的流程，你可以看到Agent如何：

1. 理解复杂任务 （分析异常）。
2. 主动查询知识库（RAG） 来明确规则。
3. 调用工具（MCP） 查询真实数据。
4. 进行分析判断 。
5. 再次调用工具 执行操作（发送通知）。
6. 生成结构化的最终报告 。

7. 企业级改造的关键考量与最佳实践

将上述Demo扩展到真实的大厂复杂项目，还需要考虑以下关键点：

7.1 架构设计最佳实践

1. 工具网关与鉴权 ：不要允许Agent直接访问内部服务的原始API。应建立一个 工具网关 ，所有工具调用都通过网关转发。网关负责统一的身份认证、权限校验、参数过滤、限流和审计日志。
2. 分层Agent设计 ：
   • 路由Agent ：接收用户请求，判断意图，分发给不同的专业Agent。
   • 专业Agent ：如订单分析Agent、客服Agent、报表生成Agent等，各司其职。
   • 工具执行层 ：封装好的各类MCP Server，提供原子能力。
3. RAG优化 ：
   • 多路召回 ：结合关键词检索（如Elasticsearch）和向量检索，提高召回率。
   • 重排序（Rerank） ：使用更精细的模型对检索结果进行重排序，提升精度。
   • 元数据过滤 ：为知识片段添加来源、部门、有效期等元数据，检索时进行过滤。
4. Prompt工程与管理 ：将Agent的提示词模板化、版本化，存储在配置中心，便于迭代和A/B测试。

7.2 安全与权限

1. 最小权限原则 ：每个Agent或工具只能访问其完成任务所必需的数据和API。
2. 输入输出净化 ：对所有用户输入和工具返回的内容进行严格的清洗和校验，防止Prompt注入、SQL注入等攻击。
3. 审计与溯源 ：记录每一次Agent的思考过程、工具调用详情、输入输出，实现全链路可追溯。
4. 人工审核环 ：对于高风险操作（如资金转账、数据删除），设计必须由人工点击确认的环节。

7.3 性能与稳定性

1. 异步与非阻塞 ：Agent的思考、工具调用（尤其是网络IO）应设计为异步，避免阻塞主线程。
2. 超时与重试 ：为每个工具调用设置合理的超时和重试机制。
3. 流式输出 ：对于耗时的复杂任务，采用流式输出（Streaming）逐步返回结果，提升用户体验。
4. 降级方案 ：当大模型服务或关键工具不可用时，应有备用的规则引擎或人工流程作为降级方案。

7.4 运维与监控

1. 成本监控 ：密切监控大模型API的Token消耗和费用，设置预算告警。
2. 效果评估 ：建立评估体系，定期用测试集评估Agent回答的准确率、工具调用的正确率。
3. 知识库更新 ：建立知识文档的更新、审核、向量化发布的自动化流水线。
4. 工具健康检查 ：对所有MCP Server进行定期健康检查，确保服务可用。

8. 常见问题排查清单

在实际落地过程中，你可能会遇到以下问题：

问题现象	可能原因	排查步骤与解决方案
Agent不调用工具，直接回答	1. Prompt指令不清晰。 2. 工具描述（ description ）不够准确。 3. 模型能力不足（如使用gpt-3.5-turbo）。	1. 强化Prompt中关于“必须使用工具”的指令。 2. 优化工具描述，明确使用场景和输入格式。 3. 升级到gpt-4或gpt-4o等工具调用能力更强的模型。
工具调用参数错误	1. Agent未能正确解析用户意图。 2. 工具的参数Schema（ args_schema ）定义有歧义。	1. 在Prompt中提供更详细的示例。 2. 使用Pydantic严格定义参数类型和描述，确保清晰无歧义。 3. 在工具函数内部增加更健壮的参数校验和错误提示。
RAG检索结果不相关	1. 文本分割策略不合理（块太大或太小）。 2. 嵌入模型不适合领域数据。 3. 检索时未使用元数据过滤。	1. 调整 chunk_size 和 chunk_overlap ，尝试不同的分割符。 2. 尝试不同的嵌入模型（如 text-embedding-3-large ）或微调领域模型。 3. 为文档添加元数据，并在检索时利用 filter 参数。
内部工具调用超时或失败	1. 网络问题或服务宕机。 2. 工具网关鉴权失败。 3. 请求参数格式错误。	1. 检查工具服务状态和网络连通性。 2. 检查网关日志，确认Token或权限是否正确。 3. 在工具封装层打印详细的请求和响应日志，便于调试。
Agent陷入循环或迭代次数过多	1. 任务过于复杂，超出Agent规划能力。 2. 工具返回的结果未能让Agent满足停止条件。	1. 设置 max_iterations （如10）进行硬性限制。 2. 优化Prompt，明确给出任务完成的判断标准。 3. 设计更细粒度的工具，或将复杂任务拆解后由路由Agent分发给多个子Agent。
回答包含幻觉或与知识库矛盾	1. RAG检索到的上下文不够或噪声太大。 2. 模型过度依赖自身知识，忽略了提供的上下文。	1. 在Prompt中强制要求模型“严格基于提供的上下文回答”，并设置惩罚。 2. 改进检索质量，提高相关片段的排名权重。 3. 采用“引用”格式，让模型在回答时注明依据的来源片段。

这套 Agent + RAG + MCP 的方案，为我们处理复杂业务逻辑与AI能力结合提供了清晰的架构范式。它不是一个银弹，而是一个需要精心设计和持续迭代的工程体系。建议从一个小而具体的业务场景开始试点，验证技术路径和业务价值，再逐步推广到更复杂的流程中。

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度