结果交付：企业级LLM MCP RAG Agent融合架构正在重构AI基建标准！ - 链载Ai

#mcp_rag_server.pyimportasyncioimportjsonimporthashlibfromtypingimportDict,Any,Optional,ListfrompathlibimportPathimportloggingfrommcp.server.fastmcpimportFastMCPfromllama_index.coreimportVectorStoreIndex,Document,Settingsfromllama_index.core.node_parserimportSentenceSplitterfromllama_index.embeddings.openaiimportOpenAIEmbeddingfromllama_index.llms.openaiimportOpenAIfromllama_index.readers.fileimportPDFReader,CSVReader#配置日志logging.basicConfig(level=logging.INFO)logger=logging.getLogger(__name__)classRAGServer:def__init__(self):self.app=FastMCP("RAG-Server")self.indicesict[str,VectorStoreIndex]={}self.document_cacheict[str,List[Document]]={}self.config=self._load_config()#初始化LlamaIndex设置Settings.llm=OpenAI(model="gpt-4o-mini")Settings.embed_model=OpenAIEmbedding(model="text-embedding-3-small")self._register_tools()def_load_config(self)->Dict[str,Any]:"""加载配置文件"""config_path=Path("doc_config.json")ifconfig_path.exists():withopen(config_path,'r',encoding='utf-8')asf:returnjson.load(f)return{"default_chunk_size":1024,"default_chunk_overlap":200}def_get_document_hash(self,file_path:str,chunk_size:int,chunk_overlap:int)->str:"""计算文档+参数的哈希值，用于缓存判断"""file_stat=Path(file_path).stat()content=f"{file_path}_{file_stat.st_size}_{file_stat.st_mtime}_{chunk_size}_{chunk_overlap}"returnhashlib.md5(content.encode()).hexdigest()def_parse_documents(self,file_path:str,chunk_size:int,chunk_overlap:int)->List[Document]:"""解析文档，支持智能缓存"""doc_hash=self._get_document_hash(file_path,chunk_size,chunk_overlap)#检查缓存ifdoc_hashinself.document_cache:logger.info(f"使用缓存文档:{file_path}")returnself.document_cache[doc_hash]#根据文件类型选择解析器file_path_obj=Path(file_path)iffile_path_obj.suffix.lower()=='.pdf':reader=PDFReader()eliffile_path_obj.suffix.lower()=='.csv':reader=CSVReader()else:raiseValueError(f"不支持的文件格式:{file_path_obj.suffix}")#解析文档documents=reader.load_data(file_path)#分块处理splitter=SentenceSplitter(chunk_size=chunk_size,chunk_overlap=chunk_overlap)nodes=splitter.get_nodes_from_documents(documents)processed_docs=[Document(text=node.text,metadata=node.metadata)fornodeinnodes]#缓存结果self.document_cache[doc_hash]=processed_docslogger.info(f"文档解析完成:{file_path},分块数:{len(processed_docs)}")returnprocessed_docsdef_register_tools(self):"""注册MCP工具"""@self.app.tool()defcreate_vector_index(file_path:str,index_name:str,chunk_size:int=None,chunk_overlap:int=None,force_recreate:bool=False)->str:"""创建向量索引Args:file_path:文档路径index_name:索引名称chunk_size:分块大小chunk_overlap:分块重叠force_recreate:强制重建"""try:#使用默认配置chunk_size=chunk_sizeorself.config["default_chunk_size"]chunk_overlap=chunk_overlaporself.config["default_chunk_overlap"]#检查是否需要重建ifnotforce_recreateandindex_nameinself.indices:returnf"索引{index_name}已存在，使用force_recreate=True强制重建"#解析文档documents=self._parse_documents(file_path,chunk_size,chunk_overlap)#创建索引index=VectorStoreIndex.from_documents(documents)self.indices[index_name]=indexreturnf"成功创建索引{index_name}，包含{len(documents)}个文档块"exceptExceptionase:logger.error(f"创建索引失败:{str(e)}")returnf"创建索引失败:{str(e)}"@self.app.tool()defquery_document(index_name:str,query:str,top_k:int=5)->str:"""查询文档Args:index_name:索引名称query:查询问题top_k:返回结果数量"""try:ifindex_namenotinself.indices:returnf"索引{index_name}不存在，请先创建索引"#执行查询query_engine=self.indices[index_name].as_query_engine(similarity_top_k=top_k)response=query_engine.query(query)returnstr(response)exceptExceptionase:logger.error(f"查询失败:{str(e)}")returnf"查询失败:{str(e)}"@self.app.tool()defget_document_summary(index_name:str,summary_type:str="brief")->str:"""获取文档摘要Args:index_name:索引名称summary_type:摘要类型(brief/detailed)"""try:ifindex_namenotinself.indices:returnf"索引{index_name}不存在"#生成摘要summary_engine=self.indices[index_name].as_query_engine()ifsummary_type=="brief":query="请用3-5句话概括这个文档的主要内容"else:query="请详细总结这个文档的核心观点和关键信息"response=summary_engine.query(query)returnstr(response)exceptExceptionase:logger.error(f"生成摘要失败:{str(e)}")returnf"生成摘要失败:{str(e)}"@self.app.tool()deflist_indices()->str:"""列出所有可用的索引"""ifnotself.indices:return"当前没有可用的索引"index_info=[]forname,indexinself.indices.items():doc_count=len(index.docstore.docs)index_info.append(f"-{name}:{doc_count}个文档块")return"可用索引：\n"+"\n".join(index_info)#服务端启动脚本asyncdefmain():server=RAGServer()awaitserver.app.run()if__name__=="__main__":asyncio.run(main())

# mcp_rag_client.pyimportasyncioimportjsonfromtypingimportDict,List,AnyfrompathlibimportPathfromlanggraph.graphimportGraph, ENDfromlanggraph.prebuiltimportToolExecutorfromlangchain.agentsimportAgentExecutorfromlangchain.schemaimportSystemMessage, HumanMessagefromlangchain_openaiimportChatOpenAIclassRAGAgent: def__init__(self, mcp_server_url:str="http://localhost:8000"):    self.mcp_server_url = mcp_server_url    self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)    self.tools = self._load_mcp_tools()    self.tool_executor = ToolExecutor(self.tools)    self.config = self._load_config()      # 构建工作流图    self.graph = self._build_workflow()  def_load_config(self) ->Dict[str,Any]:   """加载MCP配置"""    config_path = Path("mcp_config.json")   ifconfig_path.exists():     withopen(config_path,'r', encoding='utf-8')asf:       returnjson.load(f)   return{"available_indices": [],"document_descriptions": {}}  def_load_mcp_tools(self) ->List[Any]:   """从MCP服务端加载工具"""   # 这里简化处理，实际实现需要通过MCP协议获取工具列表   # 返回模拟的工具列表   return[]  def_build_workflow(self) -> Graph:   """构建LangGraph工作流"""    workflow = Graph()      # 添加节点    workflow.add_node("planner", self.planning_node)    workflow.add_node("executor", self.execution_node)    workflow.add_node("reviewer", self.review_node)      # 添加边    workflow.add_edge("planner","executor")    workflow.add_edge("executor","reviewer")      # 条件边    workflow.add_conditional_edges(     "reviewer",      self.should_continue,      {       "continue":"planner",       "end": END      }    )      # 设置入口    workflow.set_entry_point("planner")      returnworkflow.compile()  defplanning_node(self, stateict[str,Any]) ->Dict[str,Any]:   """任务规划节点"""    messages = state.get("messages", [])      # 构建系统提示，包含文档感知信息    system_prompt = self._build_system_prompt()      # 调用LLM进行规划    planning_messages = [      SystemMessage(content=system_prompt),      HumanMessage(content=f"请分析以下任务并制定执行计划：{messages[-1].content}")    ]       response = self.llm.invoke(planning_messages)      # 解析规划结果    plan = self._parse_plan(response.content)       state["current_plan"] = plan    state["plan_step"] =0      returnstate  defexecution_node(self, stateict[str,Any]) ->Dict[str,Any]:   """任务执行节点"""    plan = state.get("current_plan", [])    step = state.get("plan_step",0)      ifstep >=len(plan):      state["execution_complete"] =True     returnstate      # 执行当前步骤    current_step = plan[step]    result = self._execute_step(current_step)      # 保存执行结果   if"execution_results"notinstate:      state["execution_results"] = []    state["execution_results"].append(result)      returnstate  defreview_node(self, stateict[str,Any]) ->Dict[str,Any]:   """结果评审节点"""    results = state.get("execution_results", [])      # 评估执行结果    review_prompt =f"""    请评估以下执行结果的质量和完整性：       执行结果：{results}       判断是否需要进一步执行或调整计划。    """       response = self.llm.invoke([HumanMessage(content=review_prompt)])      # 解析评审结果    state["review_result"] = response.content    state["needs_continue"] ="需要继续"inresponse.content      returnstate  defshould_continue(self, stateict[str,Any]) ->str:   """判断是否继续执行"""   ifstate.get("execution_complete",False):     return"end"      ifstate.get("needs_continue",False):      state["plan_step"] +=1     return"continue"      return"end"  def_build_system_prompt(self) ->str:   """构建包含文档感知信息的系统提示"""    available_indices = self.config.get("available_indices", [])    doc_descriptions = self.config.get("document_descriptions", {})       prompt ="""    你是一个专业的文档分析助手，具备以下能力：    1. 文档查询和分析    2. 多文档对比分析    3. 文档摘要生成    4. 索引管理    可用的文档索引：    """      forindex_nameinavailable_indices:      description = doc_descriptions.get(index_name,"无描述")      prompt +=f"-{index_name}:{description}\n"       prompt +="""       可用工具：    - create_vector_index: 创建文档向量索引    - query_document: 查询特定文档    - get_document_summary: 获取文档摘要    - list_indices: 列出所有索引       请根据用户需求，智能选择合适的工具和索引进行任务执行。    """      returnprompt  def_parse_plan(self, plan_text:str) ->List[Dict[str,Any]]:   """解析执行计划"""   # 简化实现，实际需要更复杂的解析逻辑    steps = []    lines = plan_text.split('\n')      forlineinlines:     ifline.strip().startswith('-')orline.strip().startswith('1.'):        steps.append({         "action": line.strip(),         "tool":"query_document", # 示例         "parameters": {}        })      returnsteps  def_execute_step(self, stepict[str,Any]) ->str:   """执行单个步骤"""   # 这里应该调用MCP工具   # 简化实现   returnf"执行步骤：{step['action']}"  asyncdefrun(self, query:str) ->str:   """运行Agent"""    initial_state = {     "messages": [HumanMessage(content=query)],     "execution_results": [],     "execution_complete":False    }      # 执行工作流    final_state =awaitself.graph.ainvoke(initial_state)      # 生成最终答案    results = final_state.get("execution_results", [])       final_answer =f"""    任务执行完成！       执行结果：   {chr(10).join(results)}       如需更详细的分析，请告诉我具体要求。    """      returnfinal_answer# 客户端启动脚本asyncdefmain():  agent = RAGAgent()  # 示例查询  queries = [   "帮我分析北京和上海的税收政策差异",   "为我总结最新的AI发展报告",   "创建一个关于市场分析的新索引"  ]  forqueryinqueries:   print(f"\n查询:{query}")    result =awaitagent.run(query)   print(f"结果:{result}")if__name__ =="__main__":  asyncio.run(main())

#mcp_config.json{"server_url":"http://localhost:8000","available_indices":["tax-beijing","tax-shanghai","ai-report-2025","market-analysis"],"document_descriptions":{"tax-beijing":"北京市税收政策文件集合","tax-shanghai":"上海市税收政策文件集合","ai-report-2025":"2025年人工智能发展报告","market-analysis":"市场分析相关文档"},"tools_permissions":{"create_vector_index":true,"query_document":true,"get_document_summary":true,"list_indices":true}}#doc_config.json{"default_chunk_size":1024,"default_chunk_overlap":200,"supported_formats":["pdf","csv","txt","docx"],"embedding_model":"text-embedding-3-small","llm_model":"gpt-4o-mini","max_cache_size":1000}

[2025-07-1210:30:01]开始任务规划...[2025-07-1210:30:02]检测到多文档对比任务[2025-07-1210:30:03]规划步骤：1.查询北京税收政策2.查询上海税收政策3.执行对比分析4.生成综合报告[2025-07-1210:30:04]执行步骤1:query_document(index_name="tax-beijing",query="税收政策概述")[2025-07-1210:30:08]北京政策查询完成，获得5个相关文档块[2025-07-1210:30:09]执行步骤2:query_document(index_name="tax-shanghai",query="税收政策概述")[2025-07-1210:30:13]上海政策查询完成，获得4个相关文档块[2025-07-1210:30:14]执行步骤3:对比分析[2025-07-1210:30:18]生成对比报告完成[2025-07-1210:30:19]任务执行完成！

#Agent自动识别文件类型和创建索引awaitagent.run("创建财务报告索引")#执行日志[INFO]检测到文件:financial_report_2025.pdf[INFO]计算文档哈希:a1b2c3d4e5f6...[INFO]开始创建索引:financial-2025[INFO]文档分块完成:156个块[INFO]向量索引创建完成[INFO]生成文档摘要...

def_get_document_hash(self,file_path:str,chunk_size:int,chunk_overlap:int)->str:"""智能缓存的核心：文档内容+参数联合哈希"""file_stat=Path(file_path).stat()#文件路径+大小+修改时间+分块参数=唯一标识content=f"{file_path}_{file_stat.st_size}_{file_stat.st_mtime}_{chunk_size}_{chunk_overlap}"returnhashlib.md5(content.encode()).hexdigest()

#针对不同文档类型的优化参数DOCUMENT_CONFIGS={'financial_report':{'chunk_size':1500,'chunk_overlap':300,'reason':'财务报告需要保持数字和表格的完整性'},'legal_document':{'chunk_size':800,'chunk_overlap':150,'reason':'法律文档需要精确的条款边界'},'research_paper':{'chunk_size':1200,'chunk_overlap':200,'reason':'学术论文需要保持逻辑段落的完整性'}}

asyncdef_discover_tools(self)->List[str]:"""动态发现MCP服务端的可用工具"""#通过MCP协议获取工具列表response=awaitself.mcp_client.list_tools()return[tool['name']fortoolinresponse['tools']]

#创建虚拟环境python-mvenvmcp_rag_envsourcemcp_rag_env/bin/activate#Linux/Mac#mcp_rag_env\Scripts\activate#Windows

#安装依赖pipinstall-rrequirements.txt#requirements.txtfastapi==0.104.1uvicorn==0.24.0llama-index==0.9.15langgraph==0.0.25langchain==0.1.0langchain-openai==0.0.5mcp-server==0.1.0pymupdf==1.23.8pandas==2.1.4numpy==1.24.3

#设置OpenAIAPI密钥exportOPENAI_API_KEY="your-api-key-here"

#创建配置文件mkdirconfigcpmcp_config.jsonconfig/cpdoc_config.jsonconfig/3.启动服务#启动RAG服务端pythonmcp_rag_server.py&#启动Agent客户端pythonmcp_rag_client.py4.Docker部署#DockerfileFROMpython:3.11-slimWORKDIR/appCOPYrequirements.txt.

RUNpipinstall-rrequirements.txtCOPY..EXPOSE8000CMD["uvicorn","mcp_rag_server:app","--host","0.0.0.0","--port","8000"]#docker-compose.ymlversion:'3.8'services:rag-server:build:.ports:-"8000:8000"environment:-OPENAI_API_KEY=${OPENAI_API_KEY}volumes:-./documents:/app/documents-./config:/app/configrag-client:build:.depends_on:-rag-serverenvironment:-MCP_SERVER_URL=http://rag-server:8000command:pythonmcp_rag_client.py

#即将支持的文档类型MULTIMODAL_SUPPORT={'image':['png','jpg','jpeg','gif'],'audio':['mp3','wav','flac'],'video':['mp4','avi','mov'],'presentation':['pptx','key'],'spreadsheet':['xlsx','numbers']}#多模态处理工具@app.tool()defcreate_multimodal_index(file_path:str,content_type:str,index_name:str)->str:"""创建多模态文档索引"""ifcontent_type=='image':#使用OCR+图像理解returnprocess_image_document(file_path,index_name)elifcontent_type=='audio':#使用语音识别returnprocess_audio_document(file_path,index_name)#...其他类型处理

classIncrementalIndexManager:def__init__(self):self.change_tracker={}deftrack_document_changes(self,file_path:str):"""追踪文档变更"""current_hash=self._get_file_hash(file_path)iffile_pathinself.change_tracker:returncurrent_hash!=self.change_tracker[file_path]returnTruedefupdate_index_incrementally(self,index_name:str,changed_filesist[str]):"""增量更新索引"""#只重建变更的文档部分forfile_pathinchanged_files:self._update_document_nodes(index_name,file_path)

#分布式索引架构classDistributedRAGServer:def__init__(self,node_id:str,cluster_nodesist[str]):self.node_id=node_idself.cluster_nodes=cluster_nodesself.shard_manager=ShardManager()defcreate_distributed_index(self,documentsist[str],index_name:str):"""创建分布式索引"""#文档分片shards=self.shard_manager.create_shards(documents)#分布式处理forshardinshards:target_node=self._select_node(shard)self._send_shard_to_node(shard,target_node)

classSmartCacheManager:def__init__(self):self.access_patterns={}self.cache_priority={}defrecord_access(self,index_name:str,query:str):"""记录访问模式"""pattern_key=f"{index_name}:{query}"self.access_patterns[pattern_key]=self.access_patterns.get(pattern_key,0)+1defoptimize_cache(self):"""基于访问模式优化缓存"""#根据访问频率调整缓存优先级forpattern,frequencyinself.access_patterns.items():iffrequency>10:#高频访问self.cache_priority[pattern]='high'

链载Ai

一、为什么选择LLM+MCP+RAG+Agent融合架构？

二、系统架构设计

1、整体架构图

2、服务端实现：RAG管道的工具化

3、客户端实现：智能任务规划

三、实际应用场景演示

场景1：多文档对比分析

场景2：智能索引管理

四、性能优化的关键技术

1. 智能缓存机制

2. 参数化分块策略

3. 动态工具发现

五、生产部署指南

1. 环境准备

2. 配置文件设置

六、未来发展方向

1. 多模态扩展

2. 增量更新机制

3. 分布式处理

4. 智能缓存优化

七、总结

1、核心优势总结：

2、适用场景：

3、技术价值：