【AI RAG知识库】09.【检索】【节点4】

掌柜智库项目(RAG)实战9. 检索数据节点实现与测试9.4 网络搜索文档 (node_web_search_mcp)文件:app/query_process/agent/nodes/node_web_search_mcp.py9.4.1 mcp的调用的准备工作MCP 服务需先在阿里云百炼平台完成开通与配置才能通过代码调用以下是完整的开通流程说明步骤1前提条件已注册阿里云账号并完成实名认证百炼 MCP 服务需实名认证后使用。步骤2开通百炼 MCP 服务的详细步骤1、进入百炼 MCP 广场打开浏览器访问百炼 MCP 服务市场链接https://bailian.console.aliyun.com/cn-beijing/?spma2c4g.11186623.0.0.5f885389KrrOsZtabmcp#/mcp-market若链接失效可通过阿里云官网→“产品”→“人工智能”→“百炼”→“MCP 广场” 进入2、搜索并选择目标 MCP 服务在 MCP 广场的搜索框中输入关键词如 “联网搜索” 在搜索结果中找到目标服务如本场景的 “联网搜索”点击服务卡片进入详情页。步骤 3开通 MCP 服务进入服务详情页后点击 “开通服务” 按钮会弹出 “开通 MCP 服务” 配置窗口需完成以下配置选择 API Key从下拉框中选择已有的DASHSCOPE_API_KEY若未创建需先在百炼控制台的 “API 密钥管理” 中生成选择部署模式推荐选择「个人 FC 资源部署」资源独立、安全隔离适合正式场景测试场景可选择「公共 FC 资源部署」共享资源启动更快选择计费模式个人FC资源部署「基础模式」按调用时长计费0.000156 元 / 秒调用后释放资源成本低「极速模式」按部署时长计费0.13 元 / 时启动速度极快冷启动≤5 毫秒适合低延迟场景选择部署地域建议选择与业务服务器同地域如 “华东 2上海”降低网络延迟确认配置后点击 “确认开通” 按钮等待百炼平台完成服务部署通常 1-2 分钟。步骤 4获取 MCP 服务地址开通完成后在服务详情页的 “调用信息” 区域复制对应的MCP 服务 SSE 地址如本场景的https://dashscope.aliyuncs.com/api/v1/mcps/WebSearch/sse后续代码中需将该地址配置到.env文件的MCP_DASHSCOPE_BASE_URL中。9.4.2 处理策略1) 获取查询词从 LangGraph 全局状态对象state中提取重写后的精准查询语句rewritten_query为标准化搜索关键词若该字段为空 / 未提取到直接终止后续流程避免无效 MCP 调用。2) 初始化 MCP 连接基于MCPServerSse创建 MCP 客户端实例配置百炼 MCP 核心连接参数后通过await search_mcp.connect()建立 SSE 流式连接连接成功返回{type: connect, success: true}。核心配置参数JSON 格式{url:百炼MCP SSE接口地址.env中MCP_DASHSCOPE_BASE_URL,headers:{Authorization:百炼/阿里云API密钥.env中OPENAI_API_KEY},timeout:300,// 客户端整体超时时间秒sse_read_timeout:300// SSE流式读取超时时间秒}3) 调用搜索工具基于已建立的 MCP 连接通过call_tool()调用百炼专属搜索工具bailian_web_search工具调用固定传参格式JSON参数不可随意修改{ tool_name: bailian_web_search, // 固定值百炼搜索工具唯一标识 arguments: { query: 步骤1提取的rewritten_query, // 必选搜索查询词 count: 5 // 可选返回结果数量默认5条建议1-10 } }4) 解析与格式化接收 MCP 流式响应提取有效数据并清洗最终封装为统一格式文档列表为后续节点提供标准化数据。① MCP 原始返回值核心有效片段SSE 流式 JSON{type:tool_call,content:[{text:{\pages\: [{\title\: \结果标题\, \url\: \结果链接\, \snippet\: \核心摘要\, \source\: \数据源\}]}}]}② 解析规则过滤出type: tool_call的响应提取content[0].text并转为 JSON 对象提取对象中pages数组遍历后仅保留title/url/snippet三个核心字段对所有字段做清洗去首尾空格、过滤空值剔除snippet为空的无效结果。③ 最终格式化结果列表嵌套字典统一格式[{title:清洗后的结果标题,url:清洗后的结果链接,snippet:清洗后的核心摘要非空}]5) 更新状态与资源清理① 资源清理无论调用成功 / 失败 / 中断均通过await search_mcp.cleanup()关闭 MCP 连接释放客户端资源避免资源泄漏。② 状态更新返回将步骤 4 格式化后的文档列表以web_search_docs为字段名更新到 LangGraph 全局状态并返回供后续节点重排序、大模型生成使用无有效结果则返回空字典。最终返回状态JSON{web_search_docs:[{title:HAK 180 烫金机官方操作手册,url:https://xxx.com/hak180/manual,snippet:HAK 180 顶部50-170mm局部烫金设置操作面板【转印参数】-【区域设置】选择顶部局部输入起始50mm、结束170mm保存生效}]}9.4.3 处理关键点百炼 MCP 官方 SDK 的核心方法connect()、call_tool()、cleanup()等均为异步函数async def而本项目中使用的 LangGraph 框架其节点函数默认采用同步调用方式invoke。由于 Python 语法限制同步函数中无法直接调用异步方法会抛出SyntaxError异常因此需要让 LangGraph 的搜索节点node_web_search_mcp以同步方式运行 MCP 的异步 API核心解决方案是使用asyncio.run()做同步 - 异步桥接通过该方法临时启动一个异步事件循环执行 MCP 的所有异步代码执行完成后自动关闭循环回到同步逻辑这是 Python 中同步代码调用异步代码的标准方案。9.4.4 代码实现步骤1 准备和环境需安装百炼 MCP 官方 SDK 和 Python 异步相关依赖执行以下命令# 核心依赖百炼MCP SDKopenai-agents uv add openai-agents # 其他基础依赖若未安装langgraph、requests、python-dotenv uv add langgraph requests python-dotenv配置文件和加载环境变量配置.env 文件在项目根目录创建.env文件添加以下配置替换为自身的百炼 API 密钥# 百炼MCP WebSearch的SSE接口地址固定值无需修改 MCP_DASHSCOPE_BASE_URLhttps://dashscope.aliyuncs.com/api/v1/mcps/WebSearch/sse # 你的阿里云/百炼平台API密钥从百炼控制台获取必填 OPENAI_API_KEYyour_aliyun_bailian_api_key定义读取配置文件位置app/config/bailian_mcp_config.py# 导入核心依赖数据类、环境变量读取、路径处理fromdataclassesimportdataclassimportosfromdotenvimportload_dotenv load_dotenv()# 定义mcp的服务配置dataclassclassMcpConfig:mcp_base_url:strapi_key:strmcp_configMcpConfig(mcp_base_urlos.getenv(MCP_DASHSCOPE_BASE_URL),api_keyos.getenv(OPENAI_API_KEY))步骤2导入基础依赖importsysimportjsonimportasynciofromapp.utils.task_utilsimportadd_done_task,add_running_taskfromapp.conf.bailian_mcp_configimportmcp_configfromagents.mcpimportMCPServerSsefromapp.core.loggerimportlogger步骤3定义mcp网络访问工具asyncdefmcp_call(query): 异步调用百炼MCP搜索服务的核心函数。 该函数负责初始化MCP客户端建立SSE连接调用远程工具并返回原始结果。 :param query: 搜索查询词通常是经过改写后的精准Query :return: MCP返回的原始结果对象 (包含 content, isError 等字段) # # 初始化百炼MCP SSE客户端# ----------------------------------------------------------------------------------# MCPServerSse 是一个基于 SSE (Server-Sent Events) 协议的 MCP 客户端实现。# 它的作用是连接到阿里云百炼提供的 MCP 服务端点从而让我们可以像调用本地函数一样调用远程工具。## 参数解释# name: 客户端名称用于日志标识方便调试。# params: 连接配置字典# - url: MCP 服务的 SSE 接口地址 (例如: .../mcps/WebSearch/sse)# - headers: HTTP 请求头必须包含 Authorization 字段传入 API Key 进行鉴权。# - timeout: 连接建立和整体请求的超时时间。# - sse_read_timeout: 读取 SSE 事件流的超时时间防止流中断导致挂起。# search_mcpMCPServerSse(namesearch_mcp,params{url:mcp_config.mcp_base_url,headers:{Authorization:mcp_config.api_key},timeout:300,sse_read_timeout:300})try:logger.info(f[MCP] 正在连接百炼 WebSearch 服务:{mcp_config.mcp_base_url})# 建立与MCP服务的SSE连接异步方法需awaitawaitsearch_mcp.connect()logger.info(f[MCP] 连接成功正在调用工具 bailian_web_search 查询:{query})# 调用百炼MCP的搜索工具核心步骤# tool_name: bailian_web_search 是百炼官方定义的工具名称# arguments: 工具所需的参数这里需要 query (查询词) 和 count (返回数量)resultawaitsearch_mcp.call_tool(tool_namebailian_web_search,arguments{query:query,count:5})logger.info([MCP] 工具调用完成已获取返回结果)returnresultexceptExceptionase:logger.error(f[MCP] 调用过程中发生异常:{e},exc_infoTrue)returnNonefinally:# 无论调用成功/失败最终都关闭MCP连接释放资源异步方法awaitsearch_mcp.cleanup()步骤4主流程编写defnode_web_search_mcp(state): LangGraph同步节点函数处理MCP搜索逻辑作为整个搜索流程的入口。 该节点会调用 mcp_call 异步函数获取搜索结果并将其解析为结构化数据存储到 state 中。 :param state: LangGraph的全局状态对象包含 session_id, rewritten_query 等信息 :return: 字典包含结构化的搜索结果 web_search_docs供后续节点使用 logger.info(---node_web_search_mcp 开始处理---)# 1. 标记任务开始add_running_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))# 2. 获取查询词querystate.get(rewritten_query,)ifnotquery:# 尝试回退到原始查询querystate.get(original_query,)docs[]# 3. 执行搜索ifquery:try:# 同步-异步桥接通过asyncio.run()执行异步的mcp_call函数logger.info(f启动异步 MCP 调用Query:{query})# # MCP 返回结果格式解析说明# ----------------------------------------------------------------------# result 是一个 CallToolResult 对象 (定义在 agents.mcp.types 中)# result.content 是一个 TextContent 对象的列表通常只有一项# result.content[0].text 是一个 JSON 字符串包含实际的搜索结果## 示例数据结构# result.content[0].text # {# pages: [# {# title: HAK 180 烫金机使用手册,# url: http://example.com/manual,# snippet: 在出厂默认状态下若想设置局部转印...# },# ...# ]# }# # resultasyncio.run(mcp_call(query))# 4. 解析结果ifresultandnotresult.isErrorandresult.content:# 解析MCP原始结果提取文本内容并转为JSON对象# result.content 通常是一个列表第一项包含文本结果raw_textresult.content[0].texttry:datajson.loads(raw_text)pagesdata.get(pages)or[]logger.info(fMCP 返回原始页面数量:{len(pages)})# 遍历结果统一封装为结构化格式foriteminpages:snippet(item.get(snippet)or).strip()url(item.get(url)or).strip()title(item.get(title)or).strip()# 过滤无核心摘要的结果ifnotsnippet:continuedocs.append({title:title,url:url,snippet:snippet})exceptjson.JSONDecodeError:logger.error(fMCP 返回结果解析 JSON 失败:{raw_text[:100]}...)else:ifresultandresult.isError:logger.error(fMCP 返回错误:{result})else:logger.warning(MCP 返回结果为空或无效)logger.info(f结构化搜索结果数量:{len(docs)})exceptExceptionase:logger.error(fMCP 搜索节点执行异常:{e},exc_infoTrue)else:logger.warning(查询词为空跳过 MCP 搜索)# 5. 标记任务结束add_done_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))logger.info(---node_web_search_mcp 处理结束---)# 若有有效搜索结果返回结果供后续节点使用无则返回空字典ifdocs:return{web_search_docs:docs}return{}9.4.5 主流程测试if__name____main__:# 测试代码单独运行该文件时验证MCP搜索功能是否正常print(\n*50)print( 启动 node_web_search_mcp 本地测试)print(*50)test_state{session_id:test_mcp_session,rewritten_query:HAK 180 在出厂默认状态下若想在纸张上只把烫金膜转印到顶部 50 mm–170 mm 的局部区域应在操作面板上如何设置,is_stream:False}try:# 调用MCP搜索节点函数执行测试result_statenode_web_search_mcp(test_state)print(\n*50)print( 测试结果摘要:)search_resultsresult_state.get(web_search_docs,[])print(f搜索结果数量:{len(search_results)})ifsearch_results:print(首条结果预览:)print(json.dumps(search_results[0],indent2,ensure_asciiFalse))else:print(未获取到搜索结果)print(*50)exceptExceptionase:logger.exception(f测试运行期间发生未捕获异常:{e})