企业级AI应用实战：Agent、RAG与MCP技术栈深度集成指南

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

最近在参与公司一个核心系统的智能化改造，发现单纯调用大模型 API 的效果远不如预期。面对海量的内部文档、复杂的业务规则和频繁变更的代码库，模型经常“一本正经地胡说八道”，或者因为无法调用内部工具而“纸上谈兵”。经过一番探索和实战，我们最终确定了Agent（智能体）、RAG（检索增强生成）和 MCP（模型上下文协议）三者结合的技术路线，成功将 AI 能力深度、稳定地集成到了现有复杂系统中。

这篇文章，我将以一个“企业内部知识问答与自动化助手”的改造项目为蓝本，完整拆解这套企业级方案的落地过程。无论你是正在为团队寻找 AI 落地切入点的技术负责人，还是希望将 AI 能力集成到现有产品中的开发者，都能从本文获得从架构设计、技术选型到代码实操的全流程指南。我们将避开那些“Hello World”式的简单 demo，直接深入企业级应用面临的核心挑战与解决方案。

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

在讨论具体方案前，我们必须先理解这三个技术为何能成为解决企业复杂问题的“黄金三角”。

1.1 RAG：为模型注入“长期记忆”与“事实依据”大语言模型（LLM）拥有强大的泛化与推理能力，但其知识受限于训练数据，且存在“幻觉”问题。对于企业而言，大量的知识存在于非结构化的文档（如产品手册、设计稿、会议纪要）、结构化的数据库（如客户信息、订单数据）以及代码仓库中。RAG 的核心思想是：在回答用户问题前，先从这些外部知识源中检索出最相关的信息片段，并将其作为上下文提供给 LLM，让 LLM 基于这些“事实”来生成答案。这极大地提升了回答的准确性和专业性，并保证了信息源的时效性。

1.2 Agent：赋予模型“行动能力”与“复杂任务拆解”如果说 RAG 解决了“知识”问题，那么 Agent 则解决了“行动”问题。一个 Agent 是一个能够感知环境、进行决策并执行动作的智能体。在 AI 语境下，它通常由 LLM 作为“大脑”，负责规划、决策和反思。Agent 可以根据用户目标，自主调用各种工具（Tool），例如执行一段代码、查询数据库、调用一个 API 接口、发送邮件等。它能将复杂的用户请求（如“分析上周的销售数据并给Top 10客户发送感谢邮件”）拆解成一系列可执行的子任务，并协调完成。

1.3 MCP：构建模型与工具之间的“标准插座”这是让 Agent 在企业环境中真正发挥威力的关键。一个企业有成千上万的内外部工具和服务，如果每个工具都需要为不同的 AI 模型（如 OpenAI GPT, Claude, 本地模型）单独适配一套调用逻辑，那将是一场集成噩梦。MCP（Model Context Protocol）由 Anthropic 提出，旨在标准化 LLM 与外部工具、数据源之间的通信方式。你可以将 MCP Server 理解为各种资源和工具的标准化适配器，而 MCP Client（通常是 AI 应用或 Agent 框架）则通过统一的协议来发现和调用这些工具。这实现了工具与模型的解耦，大幅降低了集成复杂度。

三者关系简述：

• RAG负责提供精准的知识（What）。
• Agent负责制定和执行计划（How）。
• MCP负责提供标准化的工具接入（With What）。

一个强大的企业级 AI 应用，往往是：用户提出需求 -> Agent 进行规划 -> 通过 RAG 检索必要知识 -> 通过 MCP 调用工具执行 -> 整合结果返回给用户。

2. 环境准备与项目架构设计

在开始编码前，明确我们的技术栈和项目目标。

2.1 项目目标：构建企业级智能问答与自动化助手

• 功能1（知识问答）：能基于企业内部文档（Markdown、PDF、Word）、Confluence/Wiki页面、代码库回答技术、产品、流程相关问题。
• 功能2（自动化执行）：能根据自然语言指令，执行如查询数据库、创建 JIRA Ticket、发送 Slack 通知、生成数据分析报告等任务。
• 要求：响应准确、有据可查、执行可靠、易于扩展新工具。

2.2 技术栈选型

• LLM 核心：OpenAI GPT-4o / Claude 3.5 Sonnet API。也可搭配本地模型（如 Qwen2.5、DeepSeek）用于特定场景。
• Agent 框架：LangChain或LangGraph。它们提供了成熟的 Agent、Tool、Chain 抽象，生态丰富，本文以 LangChain 为例。
• RAG 核心：
  • 向量数据库：Chroma（轻量，开发友好）或Milvus/Weaviate（生产级，分布式）。
  • 文本嵌入模型：OpenAItext-embedding-3-small或开源模型BGE-M3。
  • 文档加载与切分：LangChain 的DocumentLoader和RecursiveCharacterTextSplitter。
• MCP 实现：使用MCP SDK将内部工具封装成 MCP Server。AI 应用作为 MCP Client 连接这些 Server。
• 后端/编排层：FastAPI或Spring Boot，用于构建主服务、管理会话和流程。
• 前端：简单的 Web 聊天界面（可用 Gradio, Streamlit 快速搭建）或集成到现有企业 IM（如 Slack、钉钉）。

2.3 项目目录结构一个清晰的结构是成功的一半。

enterprise-ai-assistant/ ├── app/ │ ├── main.py # FastAPI 主应用入口 │ ├── agents/ # Agent 定义 │ │ ├── __init__.py │ │ └── orchestration_agent.py # 主编排Agent │ ├── chains/ # LangChain Chains │ │ └── rag_chain.py # RAG 问答链 │ ├── tools/ # 工具定义 (兼容MCP) │ │ ├── __init__.py │ │ ├── database_tool.py # 数据库查询工具 │ │ ├── jira_tool.py # JIRA工具 │ │ └── slack_tool.py # Slack工具 │ ├── mcp_servers/ # MCP Server 实现 │ │ ├── __init__.py │ │ └── internal_tools_server.py │ ├── rag/ # RAG 核心模块 │ │ ├── __init__.py │ │ ├── vector_store.py # 向量库初始化与操作 │ │ └── ingest.py # 文档摄取管道 │ └── config.py # 配置文件 ├── data/ # 存放原始文档 │ └── knowledge_base/ ├── scripts/ # 脚本 │ └── ingest_documents.py # 文档向量化脚本 ├── requirements.txt └── .env.example

3. 核心模块拆解与实现

接下来，我们分步实现每个核心模块。

3.1 RAG 模块实现：构建企业知识库RAG 的第一步是“灌知识”。我们需要将各种格式的文档转换为向量并存储。

3.1.1 文档加载与处理我们使用 LangChain 的文档加载器。首先安装依赖：pip install langchain langchain-community chromadb pypdf python-docx markdown

创建app/rag/ingest.py：

# app/rag/ingest.py import os from typing import List from langchain_community.document_loaders import ( DirectoryLoader, PyPDFLoader, TextLoader, UnstructuredMarkdownLoader, ) from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document from .vector_store import get_vector_store def load_documents(data_dir: str) -> List[Document]: """加载指定目录下的所有文档""" documents = [] # 加载 PDF 文件 pdf_loader = DirectoryLoader( os.path.join(data_dir, "pdfs"), glob="**/*.pdf", loader_cls=PyPDFLoader, show_progress=True ) documents.extend(pdf_loader.load()) # 加载 Markdown 文件 md_loader = DirectoryLoader( os.path.join(data_dir, "mds"), glob="**/*.md", loader_cls=UnstructuredMarkdownLoader, show_progress=True ) documents.extend(md_loader.load()) # 加载文本文档 txt_loader = DirectoryLoader( os.path.join(data_dir, "txts"), glob="**/*.txt", loader_cls=TextLoader, show_progress=True ) documents.extend(txt_loader.load()) print(f"共加载 {len(documents)} 个文档片段") return documents def split_documents(documents: List[Document]) -> List[Document]: """将长文档切分成适合嵌入的片段""" text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 每个片段的最大字符数 chunk_overlap=200, # 片段之间的重叠字符数，保持上下文连贯 separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) split_docs = text_splitter.split_documents(documents) print(f"切分后得到 {len(split_docs)} 个文本块") return split_docs def ingest_to_vector_store(data_dir: str = "../data/knowledge_base"): """文档摄取主流程：加载 -> 切分 -> 向量化 -> 存储""" # 1. 加载原始文档 raw_docs = load_documents(data_dir) if not raw_docs: print("未找到任何文档，请检查 data_dir 路径。") return # 2. 切分文档 split_docs = split_documents(raw_docs) # 3. 获取向量存储实例并添加文档 vector_store = get_vector_store() # 注意：生产环境应考虑去重和增量更新，这里简单演示全量添加 # 获取所有文本的ID（例如使用内容哈希），用于后续避免重复插入 vector_store.add_documents(split_docs) print("文档已成功向量化并存储到 ChromaDB。") if __name__ == "__main__": ingest_to_vector_store()

3.1.2 向量存储初始化创建app/rag/vector_store.py：

# app/rag/vector_store.py import os from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.embeddings import CacheBackedEmbeddings from langchain.storage import LocalFileStore from app.config import settings # 初始化嵌入模型 def get_embedding_model(): # 使用 OpenAI 嵌入模型，也可替换为 HuggingFace 模型 # 例如：from langchain_huggingface import HuggingFaceEmbeddings return OpenAIEmbeddings( model="text-embedding-3-small", openai_api_key=settings.OPENAI_API_KEY ) # 可选：为嵌入添加缓存，加速重复内容的处理 def get_cached_embedder(): underlying_embeddings = get_embedding_model() store = LocalFileStore("./.cache/embeddings") # 缓存目录 cached_embedder = CacheBackedEmbeddings.from_bytes_store( underlying_embeddings, store, namespace=underlying_embeddings.model ) return cached_embedder # 获取或创建向量存储 def get_vector_store(persist_directory: str = "./chroma_db"): embedder = get_embedding_model() # 或使用 get_cached_embedder() # 如果持久化目录已存在，则加载现有库；否则创建新库 vector_store = Chroma( persist_directory=persist_directory, embedding_function=embedder, collection_name="enterprise_knowledge" ) return vector_store # 一个简单的检索函数，供后续 Chain 使用 def retrieve_docs(query: str, k: int = 4): """检索与查询最相关的 k 个文档片段""" vector_store = get_vector_store() retriever = vector_store.as_retriever(search_kwargs={"k": k}) return retriever.invoke(query)

3.1.3 构建 RAG 问答链创建app/chains/rag_chain.py：

# app/chains/rag_chain.py from langchain.chains import create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from app.rag.vector_store import get_vector_store from app.config import settings def get_rag_chain(): """创建并返回一个 RAG 问答链""" # 1. 初始化 LLM llm = ChatOpenAI( model="gpt-4o-mini", # 可根据需求调整模型 temperature=0.1, # 低温度保证答案更确定 openai_api_key=settings.OPENAI_API_KEY ) # 2. 定义系统提示词，指导模型如何利用检索到的上下文 system_prompt = ( "你是一个专业的企业知识助手。请严格根据提供的上下文信息来回答问题。\n" "如果上下文中的信息不足以回答问题，请如实告知用户你不知道，不要编造信息。\n" "请用清晰、有条理的方式组织答案。如果适用，可以引用相关的信息来源。\n" "上下文：\n{context}" ) # 3. 创建提示词模板 prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), ("human", "{input}"), ]) # 4. 创建“组合文档”链，它负责将检索到的文档和用户问题组合成最终提示 document_chain = create_stuff_documents_chain(llm, prompt) # 5. 创建检索器 vector_store = get_vector_store() retriever = vector_store.as_retriever(search_kwargs={"k": 6}) # 检索6个片段 # 6. 创建最终的检索链：检索 + 组合答案 rag_chain = create_retrieval_chain(retriever, document_chain) return rag_chain # 使用示例 if __name__ == "__main__": chain = get_rag_chain() result = chain.invoke({"input": "公司的报销流程是什么？"}) print("答案：", result["answer"]) # 可以查看来源 print("来源文档：", result["context"])

运行python scripts/ingest_documents.py完成知识库构建后，RAG 模块就准备好了。

3.2 工具模块实现：赋予 Agent “手脚”Agent 需要通过工具与外界交互。我们先实现几个典型的企业内部工具。

3.2.1 基础工具示例：数据库查询创建app/tools/database_tool.py：

# app/tools/database_tool.py from typing import Type, Optional from pydantic import BaseModel, Field from langchain.tools import BaseTool import sqlite3 # 示例用 SQLite，生产环境替换为你的数据库驱动 import pandas as pd from app.config import settings # 定义工具的输入参数模型 class DatabaseQueryInput(BaseModel): query: str = Field(description="一个清晰、可执行的 SQL 查询语句，用于查询数据库。") class DatabaseQueryTool(BaseTool): name: str = "query_database" description: str = ( "用于查询企业数据库以获取结构化数据。" "输入必须是一个有效的 SQL SELECT 语句。" "请确保查询不会修改或删除数据。" ) args_schema: Type[BaseModel] = DatabaseQueryInput def _run(self, query: str) -> str: """执行数据库查询""" try: # 生产环境请使用连接池和安全的数据库驱动（如 psycopg2, pymysql） # 这里使用 SQLite 示例 conn = sqlite3.connect(settings.DATABASE_URL) df = pd.read_sql_query(query, conn) conn.close() if df.empty: return "查询成功，但未返回任何数据。" # 将结果转换为易读的字符串格式，限制行数避免过长 result_str = df.head(10).to_string(index=False) # 只显示前10行 if len(df) > 10: result_str += f"\n... (总共 {len(df)} 行，此处显示前10行)" return result_str except Exception as e: return f"数据库查询失败，错误信息：{str(e)}。请检查 SQL 语法或表名。" async def _arun(self, query: str) -> str: """异步执行（可选）""" return self._run(query)

3.2.2 集成外部 API：JIRA 工具示例创建app/tools/jira_tool.py：

# app/tools/jira_tool.py from typing import Type, Optional from pydantic import BaseModel, Field from langchain.tools import BaseTool from jira import JIRA # 需要安装 jira 库: pip install jira from app.config import settings class CreateJiraTicketInput(BaseModel): summary: str = Field(description="JIRA Ticket 的标题/摘要。") description: str = Field(description="Ticket 的详细描述。") issue_type: str = Field(default="Task", description="问题类型，如 Bug, Task, Story。") project_key: str = Field(description="JIRA 项目键，如 PROJ。") class JiraCreateTool(BaseTool): name: str = "create_jira_ticket" description: str = ( "在指定的 JIRA 项目中创建一个新的 Ticket（任务/缺陷）。" "需要提供标题、描述、问题类型和项目键。" ) args_schema: Type[BaseModel] = CreateJiraTicketInput def _run(self, summary: str, description: str, issue_type: str = "Task", project_key: str = "PROJ") -> str: try: # 初始化 JIRA 客户端 jira = JIRA( server=settings.JIRA_SERVER, basic_auth=(settings.JIRA_USERNAME, settings.JIRA_API_TOKEN) ) # 创建 Issue issue_dict = { 'project': {'key': project_key}, 'summary': summary, 'description': description, 'issuetype': {'name': issue_type}, } new_issue = jira.create_issue(fields=issue_dict) return f"成功创建 JIRA Ticket: [{new_issue.key}]({settings.JIRA_SERVER}/browse/{new_issue.key}) - {summary}" except Exception as e: return f"创建 JIRA Ticket 失败: {str(e)}"

3.3 MCP 集成：标准化工具暴露为了让工具能被 Claude Desktop、Cursor 等支持 MCP 的客户端直接使用，或者让我们的 Agent 通过标准协议调用，我们需要将工具包装成 MCP Server。

3.3.1 安装 MCP SDKpip install mcp

3.3.2 创建 MCP Server创建app/mcp_servers/internal_tools_server.py：

# app/mcp_servers/internal_tools_server.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from app.tools.database_tool import DatabaseQueryTool from app.tools.jira_tool import JiraCreateTool # ... 导入其他工具 class InternalToolsServer: """一个简单的 MCP Server，暴露内部工具""" def __init__(self): self.tools = [ DatabaseQueryTool(), JiraCreateTool(), # 添加更多工具... ] async def list_tools(self): """返回工具列表，符合 MCP 协议""" mcp_tools = [] for tool in self.tools: mcp_tools.append({ "name": tool.name, "description": tool.description, "inputSchema": tool.args_schema.schema() if tool.args_schema else {} }) return mcp_tools async def call_tool(self, tool_name: str, arguments: dict): """调用指定工具""" for tool in self.tools: if tool.name == tool_name: try: # 根据工具定义的参数模型验证输入 if tool.args_schema: validated_args = tool.args_schema(**arguments) result = tool._run(**validated_args.dict()) else: result = tool._run(**arguments) return {"content": [{"type": "text", "text": result}]} except Exception as e: return {"content": [{"type": "text", "text": f"工具执行错误: {str(e)}"}]} return {"content": [{"type": "text", "text": f"未找到工具: {tool_name}"}]} async def main(): """启动 MCP Server (stdio 模式)""" server = InternalToolsServer() # 这里使用 stdio 通信，可以被支持 MCP 的客户端连接 async with stdio_client(StdioServerParameters(command="python", args=["-m", "app.mcp_servers.internal_tools_server"])) as (read, write): session = ClientSession(read, write) await session.initialize() # 告知客户端本 Server 提供的工具列表 tools = await server.list_tools() await session.send_list_tools(tools) # 处理客户端请求 async for message in session.listen(): if message.method == "tools/call": result = await server.call_tool(message.params.name, message.params.arguments) await session.send_result(message.id, result) if __name__ == "__main__": asyncio.run(main())

这样，任何 MCP 客户端都可以连接到这个 Server 并发现、调用query_database和create_jira_ticket等工具。

3.4 Agent 实现：组装大脑与手脚最后，我们将 RAG 链和工具组合起来，创建一个能自主决策的 Agent。

创建app/agents/orchestration_agent.py：

# app/agents/orchestration_agent.py from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from app.chains.rag_chain import get_rag_chain from app.tools.database_tool import DatabaseQueryTool from app.tools.jira_tool import JiraCreateTool from app.config import settings def get_orchestration_agent(): """创建主编排 Agent，集成 RAG 和工具调用能力""" # 1. 初始化 LLM (使用支持工具调用的模型) llm = ChatOpenAI( model="gpt-4o", # 推荐使用支持工具调用的较新模型 temperature=0, openai_api_key=settings.OPENAI_API_KEY ) # 2. 定义工具列表 tools = [ DatabaseQueryTool(), JiraCreateTool(), # 未来可以在这里添加通过 MCP Client 动态发现的工具 ] # 3. 定义系统提示词，指导 Agent 的行为 system_prompt = """你是一个强大的企业级AI助手，可以访问知识库和多种工具来帮助员工解决问题。 你的能力包括： 1. **知识问答**：你可以从企业知识库中检索信息来回答问题。 2. **数据查询**：你可以通过工具查询数据库获取实时数据。 3. **任务自动化**：你可以创建JIRA工单、发送通知等。 请遵循以下原则： - 首先，判断用户的问题是需要查询知识库，还是需要执行一个动作（使用工具），或是两者都需要。 - 如果问题涉及公司制度、产品信息、技术文档等静态知识，优先使用知识库（RAG）。 - 如果问题需要获取实时数据（如销售数据、用户统计）或执行一个操作（如创建任务），则使用相应的工具。 - 在回答时，请清晰说明你的思考过程和信息来源。 - 如果工具执行失败或知识库没有相关信息，请如实告知用户。 """ # 4. 构建 Agent 提示词模板 prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 用于记录 Agent 的思考过程 ]) # 5. 创建 Agent agent = create_openai_tools_agent(llm, tools, prompt) # 6. 创建 Agent 执行器 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 设置为 True 可以看到 Agent 的思考步骤，生产环境可关闭 handle_parsing_errors=True, # 优雅处理解析错误 max_iterations=5, # 防止 Agent 陷入无限循环 ) return agent_executor # 注意：这个 Agent 目前还没有直接集成 RAG 链。 # 一种高级模式是：将 RAG 链本身也封装成一个 Tool，让 Agent 在需要时主动调用。 # 另一种更紧密的模式是：在 Agent 的决策逻辑中，先调用 RAG，再根据 RAG 的结果决定是否使用其他工具。 # 下面我们实现一个更集成的版本： class EnhancedOrchestrationAgent: """增强版编排 Agent，内部集成 RAG 调用逻辑""" def __init__(self): self.rag_chain = get_rag_chain() self.tool_agent = get_orchestration_agent() # 上面定义的纯工具Agent def invoke(self, user_input: str) -> str: """处理用户输入的核心逻辑""" # 启发式判断：如果问题包含明显的动作动词或需要实时数据，可能更需要工具 action_keywords = ["查询", "获取", "创建", "发送", "计算", "统计", "执行"] needs_tool = any(keyword in user_input for keyword in action_keywords) # 启发式判断：如果问题明显是关于知识的 knowledge_keywords = ["是什么", "流程", "指南", "文档", "如何理解", "介绍"] needs_rag = any(keyword in user_input for keyword in knowledge_keywords) final_answer = "" # 策略1：如果既需要知识也需要动作，先获取知识再执行动作（简单串联） if needs_rag: rag_result = self.rag_chain.invoke({"input": user_input}) knowledge_context = rag_result["answer"] final_answer += f"【根据知识库】\n{knowledge_context}\n\n" # 如果知识已经回答了问题，且不需要额外动作，可以提前结束 if not needs_tool or "不知道" in knowledge_context: return final_answer.strip() # 否则，将知识作为上下文，继续询问是否需要工具执行 user_input_with_context = f"背景知识：{knowledge_context}\n用户原问题：{user_input}\n基于以上知识，是否需要执行额外操作？如果需要，请说明。" # 策略2：执行工具调用 if needs_tool: try: tool_result = self.tool_agent.invoke({"input": user_input}) final_answer += f"【执行结果】\n{tool_result['output']}" except Exception as e: final_answer += f"\n工具执行过程中出现错误：{str(e)}" return final_answer if final_answer else "我暂时无法处理这个问题，请尝试更清晰地描述您的需求。" # 使用示例 if __name__ == "__main__": agent = EnhancedOrchestrationAgent() # 测试知识类问题 answer1 = agent.invoke("公司的年假制度是怎样的？") print("问题1回答：", answer1) print("-" * 50) # 测试工具类问题 answer2 = agent.invoke("查询一下上个月销售额最高的产品是什么？") print("问题2回答：", answer2) print("-" * 50) # 测试混合问题 answer3 = agent.invoke("根据报销流程，我需要为上周的差旅创建一张JIRA报销审批单，标题是'北京差旅报销'。") print("问题3回答：", answer3)

4. 服务集成与 API 暴露

最后，我们用 FastAPI 将上述所有能力封装成统一的 API 服务。

创建app/main.py：

# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.agents.orchestration_agent import EnhancedOrchestrationAgent import uvicorn from app.config import settings app = FastAPI(title="企业级AI助手API", description="集成RAG、Agent与工具调用的智能助手") # 初始化全局 Agent（生产环境应考虑生命周期和并发） agent = None @app.on_event("startup") async def startup_event(): global agent print("正在初始化AI助手...") agent = EnhancedOrchestrationAgent() print("AI助手初始化完成。") class UserQuery(BaseModel): question: str session_id: str = None # 可用于支持多轮对话 @app.post("/ask") async def ask_question(query: UserQuery): """核心问答接口""" if not agent: raise HTTPException(status_code=503, detail="服务正在初始化，请稍后重试。") try: answer = agent.invoke(query.question) return { "success": True, "answer": answer, "session_id": query.session_id } except Exception as e: raise HTTPException(status_code=500, detail=f"处理问题时发生错误：{str(e)}") @app.get("/health") async def health_check(): return {"status": "healthy"} if __name__ == "__main__": uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

现在，运行python -m app.main即可启动服务，通过POST /ask接口进行提问。

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

将上述 demo 改造为真正稳定、可靠的企业级系统，还需要考虑以下方面：

5.1 性能与扩展性

• 向量库选择：Chroma 适合原型和中小规模，生产环境应考虑 Milvus、Weaviate、PgVector（与 PostgreSQL 集成）等支持分布式、持久化和高性能检索的方案。
• 嵌入缓存：对重复内容（如常见问题）的嵌入结果进行缓存，显著降低成本和延迟。
• 异步处理：文档摄取、向量化、工具调用等耗时操作应使用异步（Async）避免阻塞主线程。
• Agent 流式输出：对于长任务，通过 Server-Sent Events (SSE) 或 WebSocket 流式返回 Agent 的思考步骤和结果，提升用户体验。

5.2 可靠性、安全与权限

• 工具权限控制：不是所有用户都能调用所有工具。需要在 Agent 调用工具前，根据用户身份（从请求头或 Token 中解析）进行权限校验。可以在 Tool 的_run方法开头加入权限判断。
• SQL 注入防护：在数据库查询工具中，严禁直接拼接用户输入。应使用参数化查询或严格限制可查询的表和字段。更好的做法是提供特定的“数据查询”API，而不是暴露原始 SQL 能力。
• 审计与日志：记录所有用户查询、Agent 的决策过程、工具调用详情和结果。这对于问题排查、效果分析和安全审计至关重要。
• 限流与熔断：对 API 和底层 LLM 调用实施限流，防止误用或攻击导致服务过载。

5.3 RAG 效果优化

• 检索优化：
  • 混合检索：结合向量检索（语义相似度）和关键词检索（BM25），提升召回率。
  • 重排序（Re-ranking）：使用更精细的模型（如 Cohere Rerank, BGE Reranker）对初步检索结果进行重排，提升精度。
  • 元数据过滤：为文档片段添加来源、部门、更新时间等元数据，检索时进行过滤。
• 提示工程：精心设计 RAG 和 Agent 的系统提示词，明确其角色、知识边界和行为规范，这是提升效果性价比最高的方式。

5.4 部署与运维

• 容器化：使用 Docker 将应用、向量数据库、MCP Server 等分别容器化，通过 Docker Compose 或 Kubernetes 编排。
• 配置管理：所有 API Key、数据库连接串等敏感信息必须通过环境变量或配置中心（如 Apollo）管理，切勿硬编码。
• 监控与告警：监控服务健康度、响应延迟、LLM API 消耗、错误率等关键指标，并设置告警。

6. 常见问题与排查思路

这套Agent × RAG × MCP的架构，为我们应对企业复杂场景提供了清晰的路径。RAG 解决了知识准确性问题，Agent 解决了复杂任务自动化问题，而 MCP 则像胶水一样，以标准化方式将企业内部五花八门的系统连接起来，让 AI 真正成为团队的一员。

改造过程并非一蹴而就，建议从一个明确的垂直场景（如“IT 帮助台问答”或“销售数据查询”）开始，验证技术路线的可行性，再逐步扩展工具和知识库的范围。在开发过程中，持续进行效果评估和安全审计，确保 AI 应用在提升效率的同时，也足够可靠和安全。

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