当前位置： 首页 > wzjs >正文

业务平台迅速上排名网站优化

wzjs 2025/8/3 9:07:28
业务平台,迅速上排名网站优化,日照网站建设全55长,zblog仿wordpressMCP项目开发-一个简单的RAG示例 前言 前言 客户端是基于官网的例子改的，模型改成了openai库连接仅仅使用基础的RAG流程作为一个演示，包含了以下步骤 query改写搜索：使用google serper重排序：使用硅基流动的api 大模型api也使用…

MCP项目开发-一个简单的RAG示例

前言

前言

  • 客户端是基于官网的例子改的，模型改成了openai库连接
  • 仅仅使用基础的RAG流程作为一个演示，包含了以下步骤
    1. query改写
    2. 搜索：使用google serper
    3. 重排序：使用硅基流动的api
  • 大模型api也使用的硅基流动
  • 调用速度还可以

官方文档

  • mcp官方文档

    https://modelcontextprotocol.io/introduction

使用到的api注册

  • 检索：google serper

    https://serper.dev

    有2500 Credits，具体也没搞明白，好像是每次检索返回的文本数量超过10则花费2Credits,不过这个不重要，反正自己用肯定够,api在这

    ​ image-20250406165845194

    连接示例

    curl --location 'https://google.serper.dev/search' \
--header 'X-API-KEY: 你的api-key' \
--header 'Content-Type: application/json' \
--data '{"q": "你的问题","num": 2
}'

    其中q表示要检索的问题，然后num为返回的文本数,返回示例如下

    {"searchParameters": {"q": "中国首都是哪","type": "search","num": 2,"engine": "google"},"organic": [{"title": "中华人民共和国首都 - 维基百科","link": "https://zh.wikipedia.org/zh-hans/%E4%B8%AD%E5%8D%8E%E4%BA%BA%E6%B0%91%E5%85%B1%E5%92%8C%E5%9B%BD%E9%A6%96%E9%83%BD","snippet": "中华人民共和国首都位于北京市，新中国成立前夕的旧称为北平，是中共中央及中央人民政府所在地，中央四个直辖市之一，全国政治、文化、国际交往和科技创新中心，中国古都、 ...","position": 1},{"title": "中华人民共和国首都_中华人民共和国中央人民政府门户网站","link": "https://www.gov.cn/guoqing/2005-05/24/content_2615214.htm","snippet": "1949年9月27日，中国人民政治协商会议第一届全体会议一致通过中华人民共和国的国都定于北平，即日起北平改名北京。 北京，简称京，是中国共产党中央委员会、中华人民共和国 ...","position": 2}],"relatedSearches": [{"query": "中国首都是北京还是上海"},{"query": "中国首都为什么是北京"},{"query": "中国首都上海"},{"query": "中国以前的首都"},{"query": "新中国首都选址"},{"query": "中国历代首都"},{"query": "中国首都候选"},{"query": "中国首都英文"}],"credits": 1
}

    主要的就是在organic中，如果我们的num为2，则organic中返回两个dict，我们取的是没个dict中的snippet字段

  • 聊天大模型与重排序模型：硅基流动

    https://cloud.siliconflow.cn/

    注册可以看我的另一篇

    https://blog.csdn.net/hbkybkzw/article/details/145558234

    api-key如图

    ​ image-20250406165928270

服务器

环境搭建

  • 官方推荐使用uv来管理环境，先下载uv

    pip install uv#或者macos linux
curl -LsSf https://astral.sh/uv/install.sh | sh#或者windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

    之后重启终端

  • 初始化项目。名称为rag_sample

    uv init rag_sample

  • 创建虚拟环境并激活

    cd rag_sample
uv venv# windows
.venv\Scripts\activate# linux
source .venv/bin/activate

  • 安装依赖

    openai库是客户端用的，为了方便，客户端就不初始化了，直接和服务器用同一个环境

    # windows
uv add mcp[cli] aiohttp dotenv openai rich# linux
uv add "mcp[cli]" aiohttp dotenv openai rich

构建服务器

  • 在rag_sample目录下，创建 .env 文件保存为环境变量

    GOOGLE_SERPER_URL=https://google.serper.dev/search
GOOGLE_SERPER_API=
SILICONFLOW_RERANKING_BASE_URL=https://api.siliconflow.cn/v1/rerank
SILICONFLOW_CHAT_BASE_URL=https://api.siliconflow.cn/v1/chat/completions
SILICONFLOW_RERANKING_API_KEY=
RERANKING_MODEL=BAAI/bge-reranker-v2-m3
OPENAI_CHAT_MODEL=Qwen/QwQ-32B
OPENAI_BASE_URL=https://api.siliconflow.cn/v1

    将GOOGLE_SERPER_API和SILICONFLOW_RERANKING_API_KEY填为自己的

  • rag_client.py

    import mcp
import aiohttp
import asyncio
from mcp.server.fastmcp import FastMCP
from utils import query_rewrite,search_with_serper,reranking_by_bgemcp = FastMCP("rag")@mcp.tool()
async def rag_sample(query:str,doc_num:int=10,top_n:int=4) -> list:"""rag_sample:   输入query,输出检索重排序后的文档Parameters:-------------------------------------------------------------------------------------------------------------------query:               type: str              desc: 用户问题               doc_num:             type: int              desc: 检索文档数              top_n:               type: top_n            desc: 重排序后返回top_n              Returns:-------------------------------------------------------------------------------------------------------------------reranked_doc:        type: list             desc: top_n检索文档"""rewrite_query = await query_rewrite(query=query)documents = await search_with_serper(query=rewrite_query,document_num=doc_num)reranked_doc = await reranking_by_bge(query=query,documents=documents,top_n=top_n)return reranked_docasync def test():query = "MCP协议介绍"reranked_doc = await rag_sample(query=query)importrich.print(reranked_doc)if __name__ == "__main__":# asyncio.run(test())mcp.run(transport="stdio")

    test函数是用来测试的

    ​ image-20250406201419091

    我们将主要的代码放在rag_client.py，辅助函数都放在utils.py中

    主要的流程就是

    1. query_rewrite： 传入用户问题，返回改写后的问题
    2. search_with_serper： 传入改写后的问题和检索数量，返回检索到的文档信息
    3. reranking_by_bge：传入用户问题和检索到的信息，返回相似度高的top_n文档信息

    代码见下面的utils.py

  • utils.py

    import asyncio
import aiohttp
from dotenv import load_dotenv
import os#从.env文件加载环境变量
load_dotenv()# google serper
google_serper_url = os.getenv("GOOGLE_SERPER_URL")
google_serper_api = os.getenv("GOOGLE_SERPER_API")# siliconflow api
chat_base_url = os.getenv("SILICONFLOW_CHAT_BASE_URL")
reranking_base_url = os.getenv("SILICONFLOW_RERANKING_BASE_URL")
reranking_api_key = os.getenv("SILICONFLOW_RERANKING_API_KEY")
reranking_model = os.getenv("RERANKING_MODEL")# query rewrite prompt
query_rewrite_prompt = """
请根据以下规则优化用户的问题：
1. 首先判断用户query是否为问候或打招呼（例如：'你好'、'在吗'、'您好'等）。如果是问候，直接返回原文query。
2. 如果不是，对query进行优化改写，请保持原意并使改写后的query更精准更简洁,限制在15个字以内
示例：- 输入：'你好，能帮个忙吗？'输出：'你好，能帮个忙吗？'- 输入：'怎么重写句子结构？'输出：'句子结构优化步骤示例'输出时无需解释，仅返回最终结果。
"""async def fetch_search_results(url, headers, payload):"""发动请求并获取响应"""async with aiohttp.ClientSession() as session:async with session.post(url, headers=headers, json=payload) as response:assert response.status == 200response_json = await response.json()return response_json  # 返回响应jsonasync def query_rewrite(query:str):"""一个简单的query改写，直接用aiohttp了，没有使用openai"""url = chat_base_urlheaders = {'Authorization': f'Bearer {reranking_api_key}','Content-Type': 'application/json'}payload={"model": "Qwen/Qwen2.5-72B-Instruct","messages": [{"role": "system","content": query_rewrite_prompt},{"role": "user","content": query}],"max_tokens": 512,"stream":False}response_data = await fetch_search_results( url=url,headers=headers,payload=payload)rewrite_query = response_data['choices'][0]['message']['content']return rewrite_queryasync def search_with_serper(query:str,document_num:int=10):"""google serper 搜索并返回documents,形式如下[{"title1":"","link1":"","snippet1":"","position1":""},{"title2":"","link2":"","snippet2":"","position2":""},...]"""url = google_serper_urlheaders = {'X-API-KEY': google_serper_api,'Content-Type': 'application/json'}payload = {"q": query, # 要搜索的问题"num": document_num # 要所搜的文档数量}response_data = await fetch_search_results( url=url,headers=headers,payload=payload)documents = response_data['organic']return documents async def reranking_by_bge(query:str,documents:list[dict],top_n:int=4):"""调用排序模型，返回排序后的文档"""doc_to_rerank = [doc["snippet"] for doc in documents]url = reranking_base_urlheaders = {'Authorization': f'Bearer {reranking_api_key}','Content-Type': 'application/json'}payload = {"model": reranking_model,"query": query,"documents": doc_to_rerank,"top_n": top_n,"return_documents": True, # 如果是false则只返回索引，麻烦"max_chunks_per_doc": 1024,"overlap_tokens": 80}reranked_info = await fetch_search_results( url=url,headers=headers,payload=payload)reranked_doc =[item['document']['text'] for item in reranked_info['results']]return reranked_docasync def test():import richquery = "流水不争先"rewrite_query = await query_rewrite(query=query)rich.print('rewrite_query',rewrite_query)documents = await search_with_serper(query=rewrite_query,document_num=4)rich.print('documents',documents)reranked_doc = await reranking_by_bge(query=query,documents=documents,top_n=2)rich.print("reranked_doc",reranked_doc)if __name__ == "__main__":asyncio.run(test())

    test函数是用来测试的

    ​ image-20250406202012371

  • 运行 python rag_client.py以确认一切正常

app端使用(cherrystudio为例)

  • 安装cherrystudio,并配置好模型

    https://cherry-ai.com/

  • 在app端添加单个weather服务器,下面的json文件需要复制到cherrystudio中

    {"mcpServers": {"rag": {"command": "uv","args": ["--directory","C:\\Users\\孟智超\\Desktop\\mac_backups\\MCP\\rag_sample","run","rag_client.py"]}}
}

    意思是：

    1. 有一个名为“天气”的 MCP 服务器
    2. 通过运行 uv --directory C:\\Users\\孟智超\\Desktop\\mac_backups\\MCP\\rag_sample run rag_client.py 来启动它（项目的地址）

  • 将上面的json文本复制到cherrystudio

    ​ image-20250406202414304

  • 选择模型，因为mcp也借用了functioncalling，所以需要选用的模型必须支持functioncalling（tools）

    在cherrystudio中的表现就是，有绿色的小扳手

    ​ image-20250406202448479

  • 开始聊天之前开启mcp

    ​ image-20250406202556713

  • 看看使用前后的对比

    问题：什么是MCP?

    未开启前,列举了很多，但是都不是我们想要的

    ​ image-20250406202722555

    开启后，虽然在思考过程中也提到了其他无关的，但是最后还是决定调用工具rag_sample，最后得到的是我们想要的正确答案。

    ​ image-20250406202908565

客户端

代码

  • 和服务器共用一个环境（不推荐，一般重新开一个环境）

  • 代码rag_client.py

    import os
import sys
import json
import asyncio
from typing import Optional
from contextlib import AsyncExitStackfrom mcp import ClientSession,StdioServerParameters
from mcp.client.stdio import stdio_clientimport openai
from openai import OpenAI
from dotenv import load_dotenv
import traceback# 从.env文件加载环境变量
load_dotenv() # siliconflow api
openai.base_url = os.getenv("OPENAI_BASE_URL")
openai.api_key = os.getenv("SILICONFLOW_RERANKING_API_KEY")
openai_chat_model = os.getenv("OPENAI_CHAT_MODEL")class MCPClient:def __init__(self):# 初始化会话和客户端对象self.session: Optional[ClientSession] = None # 保存mcp客户端会话self.exit_stack = AsyncExitStack()self.model = openai_chat_modelself.openai = OpenAI(base_url=openai.base_url,api_key=openai.api_key)async def connect_to_server(self, server_script_path:str):"""connect_to_server:   连接到mcp服务器 Parameters:-------------------------------------------------------------------------------------------------------------------server_script_path:         type:str             desc:  server脚本地址（.py or .js）            """is_python = server_script_path.endswith('.py') # python脚本？is_js = server_script_path.endswith('.js') # node 脚本？if not (is_python or is_js):raise ValueError("Server script must be a .py or .js file")command = "python" if is_python else "node"server_params = StdioServerParameters(command=command,args=[server_script_path],env=None) # 告诉mcp客户端如何启动服务器,比如当前的 python rag_server.pystdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params) # 启动服务器进程，建立标准I/O通信管道)self.stdio,self.write = stdio_transport # 拿到读写流self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio,self.write) # 创建MCP客户端会话，与服务器交互)await self.session.initialize() # 发送初始化消息给服务器，等待服务器就绪。# 列出 mcp 服务器上可用的工具response = await self.session.list_tools() # 向 mcp 服务器请求所有已注册的工具（即@mcp.tool）tools = response.toolsprint("\nConnected to server with tools:", [tool.name for tool in tools])async def process_query(self,query:str) -> str:"""process_query:   使用大模型处理查询，调用已经注册的MCP工具 Parameters:-------------------------------------------------------------------------------------------------------------------query:               type:str               desc:用户问题                 """messages = [{"role":"user","content": query}]response = await self.session.list_tools() # 向 mcp 服务器请求所有已注册的工具（即@mcp.tool）# 转换为functional calling 的available_tools格式available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"input_schema": tool.inputSchema}} for tool in response.tools]# print(f"available_tools：\n{available_tools}")# 初始化 大模型连接(OpenAI)response = self.openai.chat.completions.create(model=self.model,max_tokens=1000,messages=messages,tools=available_tools)content = response.choices[0]print(f"content.finish_reason : {content.finish_reason}")if content.finish_reason == "tool_calls": print(f"进入functional calling")# content.finish_reason == "tool_calls"表示要调用工具# 会在content.message.tool_calls 列表中声明要用哪个函数、参数是什么tool_call = content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 取出工具名 tool_name 和参数 tool_args，执行mcp工具result = await self.session.call_tool(tool_name, tool_args)print(f"\n[Calling tool {tool_name} with args {tool_args}]\n\n")# print(f"\n\ncall_tool_result: {result}")# print(f"\n\ncall_tool_result.content[0].text:{result.content[0].text}")# print(f"i.text for i in result.content:{[i.text for i in result.content]}")# 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中messages.append(content.message.model_dump())messages.append({"role": "tool","content": f"{[i.text for i in result.content]}","tool_call_id": tool_call.id,})# 将上面的结果再次返回给大模型并产生最终结果response = self.openai.chat.completions.create(model=self.model,max_tokens=1000,messages=messages,)return response.choices[0].message.content# 如果没有调用工具则直接返回当前结果return content.message.contentasync def chat_loop(self):"""运行一个交互式聊天循环"""print("MCP 服务器已经启动!")print("输入你的问题（输入'quit'退出）.")while True:try:query = input("\n你的问题:\n").strip()if query.lower() == "quit":breakresponse = await self.process_query(query)print(f"模型回答：\n{response}")except Exception as e:tb = traceback.extract_tb(e.__traceback__)for frame in tb:print("File:", frame.filename)print("Line:", frame.lineno)print("Function:", frame.name)print("Code:", frame.line)async def cleanup(self):"""清理资源"""# 异步地关闭所有在 exit_stack 中注册的资源（包括 MCP 会话）。await self.exit_stack.aclose()async def main():"""main:   主执行逻辑，以现在的rag为例子标准IO启动需要服务器的代码作为客户端的子进程启动服务器代码：rag_server.py客户端代码：rag_client.py因为都在同一个项目下，所以启动直接为python rag_client.py rag_server.py"""if len(sys.argv) < 2: # print("Usage: python client.py <path_to_server_script>")sys.exit(1)client = MCPClient()try:print(sys.argv[1])# python rag_client.py rag_server.py ,则argv[1] = rag_client.py 为客户端代码await client.connect_to_server(sys.argv[1]) await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())

    1. connect_to_server：连接到MCP服务器

    2. process_query： 使用大模型处理查询，将MCP工具转换为OpenAI兼容的格式，允许模型决定是否调用工具，如果是，则调用已经注册的MCP工具

    3. chat_loop： 交互式聊天（只有单轮）

    4. main：主函数

      因为在客户端脚本中，启动方式是"stdio"

      if __name__ == "__main__":mcp.run(transport="stdio")

      这种适合在客户端和服务器在同一个环境中的情况下，是将服务器作为客户端的子进程

      所以其中这个客户端的命令如下

      python rag_client.py rag_server.py

测试运行

  • 启动

    python rag_client.py rag_server.py

  • 测试下

    (模型不太好触发这个tool，所以我直接加了“请调用rag_sample”)

    问题：

    请调用rag_sample,解释下最近很火的mcp

    ​ image-20250406204328355

http://www.dtcms.com/wzjs/200935.html

相关文章：

  • 帮人做视频的网站西点培训班一般要多少学费
  • wordpress竖着的分割线知乎seo排名的搜软件
  • 北京工程建设质量协会网站厦门网站建设平台
  • 什么是seo关键词盐城seo排名
  • 做网站按页面收费网站关键词优化推广哪家快
  • 男人和女人做不可描述的事情的网站成都seo招聘
  • 做网站一定需要服务器吗站长工具seo诊断
  • 专业做网站公司 前景腾讯企点客服
  • wordpress最底部搜索引擎优化百度百科
  • 河南城乡建设网站seo网站排名优化服务
  • 口碑好的网站开发公司电话app运营方案策划
  • 龙岗公司做网站能让手机流畅到爆的软件
  • 连云港网站定制开发东莞seo黑帽培训
  • 网站建设公司福州宝鸡百度seo
  • 一般通过 什么意思seo综合排名优化
  • 重庆网站制作招聘百度关键词优化推广
  • 文登 网站建设seo博客优化
  • 河南教育平台网站建设推广发布任务平台app下载
  • 北京做网站推广百度关键词优化排名
  • 金融建设网站河南制作网站
  • 做网站用win还是li百度网站排名查询工具
  • 做网站ps建立多大的画布地推怎么做最有效
  • 做色情诱导网站广东新闻今日大件事
  • 怎么做网站关键词视频seo系统教程
  • 私人怎么做彩票网站平台药品销售推广方案
  • 萍乡网站开发公司南宁seo产品优化服务
  • 温州二井建设有限公司网站潍坊百度seo公司
  • 网站建设网店名字站长之家关键词查询
  • 泗阳县住房和城乡建设局网站论文收录网站有哪些
  • 移动端网站开发教程新媒体代运营