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, 本地模型单独适配一套调用逻辑那将是一场集成噩梦。MCPModel 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.example3. 核心模块拆解与实现接下来我们分步实现每个核心模块。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_clsPyPDFLoader, show_progressTrue ) documents.extend(pdf_loader.load()) # 加载 Markdown 文件 md_loader DirectoryLoader( os.path.join(data_dir, mds), glob**/*.md, loader_clsUnstructuredMarkdownLoader, show_progressTrue ) documents.extend(md_loader.load()) # 加载文本文档 txt_loader DirectoryLoader( os.path.join(data_dir, txts), glob**/*.txt, loader_clsTextLoader, show_progressTrue ) documents.extend(txt_loader.load()) print(f共加载 {len(documents)} 个文档片段) return documents def split_documents(documents: List[Document]) - List[Document]: 将长文档切分成适合嵌入的片段 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段的最大字符数 chunk_overlap200, # 片段之间的重叠字符数保持上下文连贯 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( modeltext-embedding-3-small, openai_api_keysettings.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, namespaceunderlying_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_directorypersist_directory, embedding_functionembedder, collection_nameenterprise_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( modelgpt-4o-mini, # 可根据需求调整模型 temperature0.1, # 低温度保证答案更确定 openai_api_keysettings.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(indexFalse) # 只显示前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 集成外部 APIJIRA 工具示例创建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(descriptionJIRA Ticket 的标题/摘要。) description: str Field(descriptionTicket 的详细描述。) issue_type: str Field(defaultTask, description问题类型如 Bug, Task, Story。) project_key: str Field(descriptionJIRA 项目键如 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( serversettings.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(fieldsissue_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 mcp3.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(commandpython, 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( modelgpt-4o, # 推荐使用支持工具调用的较新模型 temperature0, openai_api_keysettings.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_nameagent_scratchpad), # 用于记录 Agent 的思考过程 ]) # 5. 创建 Agent agent create_openai_tools_agent(llm, tools, prompt) # 6. 创建 Agent 执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 设置为 True 可以看到 Agent 的思考步骤生产环境可关闭 handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5, # 防止 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_code503, detail服务正在初始化请稍后重试。) try: answer agent.invoke(query.question) return { success: True, answer: answer, session_id: query.session_id } except Exception as e: raise HTTPException(status_code500, detailf处理问题时发生错误{str(e)}) app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: uvicorn.run(app.main:app, host0.0.0.0, port8000, reloadTrue)现在运行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. 常见问题与排查思路问题现象可能原因排查步骤与解决方案RAG 回答不相关或“幻觉”1. 检索到的文档不相关。2. 提示词未强制模型基于上下文。3. 文档切分不合理上下文断裂。1. 检查检索结果在调用链中打印出retriever返回的文档看是否匹配问题。2. 强化系统提示词加入“严格基于上下文”的指令。3. 调整chunk_size和chunk_overlap或尝试按段落/标题切分。Agent 频繁调用错误工具或陷入循环1. 工具描述不清晰。2. Agent 的max_iterations设置过高。3. 系统提示词未明确决策逻辑。1. 优化工具的name和description使其更精准。2. 适当降低max_iterations如设为 5-10。3. 在提示词中明确决策流程例如“先判断问题类型再选择工具”。MCP 客户端连接 Server 失败1. MCP Server 未正确启动或崩溃。2. 通信协议或参数不匹配。3. 防火墙/端口问题。1. 检查 Server 日志确保stdio_client参数正确。2. 使用mcp包的 CLI 工具测试 Servermcp dev ./path/to/server_script.py。3. 确认客户端配置的 Server 命令路径正确。文档向量化速度慢1. 嵌入模型调用慢尤其是网络请求。2. 文档数量多、体积大。3. 未使用缓存。1. 考虑使用更快的本地嵌入模型如all-MiniLM-L6-v2。2. 实现增量更新只处理新文档。3. 启用嵌入缓存如示例中的CacheBackedEmbeddings。工具调用权限错误1. 工具内未做权限校验。2. 用户 Token 解析失败或过期。1. 在每个工具的_run方法开始处从全局请求上下文获取用户信息并进行权限判断。2. 确保 API 网关或认证中间件正确传递了用户身份信息。这套Agent × RAG × MCP的架构为我们应对企业复杂场景提供了清晰的路径。RAG 解决了知识准确性问题Agent 解决了复杂任务自动化问题而 MCP 则像胶水一样以标准化方式将企业内部五花八门的系统连接起来让 AI 真正成为团队的一员。改造过程并非一蹴而就建议从一个明确的垂直场景如“IT 帮助台问答”或“销售数据查询”开始验证技术路线的可行性再逐步扩展工具和知识库的范围。在开发过程中持续进行效果评估和安全审计确保 AI 应用在提升效率的同时也足够可靠和安全。 30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度