MCP Sampling 示例让 AI 智能体与人类协作采样引言什么是 MCP Sampling想象一下你正在开发一个智能助手它需要从海量数据中筛选出最有价值的信息。如果让它自己盲目采样可能会遗漏重要内容如果完全依赖人工又效率低下。这时候MCPModel Context Protocol的 Sampling 机制就派上用场了。MCP Sampling 是一种让 AI 模型能够根据上下文主动请求额外数据的技术。它就像给 AI 装了一个「智能抽水机」——当模型发现当前信息不足时会主动请求从外部数据源抽取更多样本同时还可以请求人类用户协助确认或补充。这种机制在需要精确决策、实时数据处理或用户个性化交互的场景中特别有用。## 核心原理MCP Sampling 的工作流程MCP Sampling 的核心是「请求-响应」模式。当 AI 模型处理任务时如果检测到信息缺口会通过 MCP 协议发送一个samplingRequest消息。这个请求包含-prompt描述需要什么数据-maxTokens期望返回的最大 token 数-stopSequences停止采样的标记-temperature采样随机性控制服务器收到请求后可以自动返回数据也可以将请求转发给人类用户进行手动采样。这种双模式设计让系统既高效又灵活。## 环境准备搭建 MCP Sampling 实验环境在开始写代码前我们需要安装必要的库。MCP 的 Python 实现已经封装好了核心功能。bashpip install mcp-python这个库提供了完整的 MCP 客户端和服务器实现包括 Sampling 相关的 API。## 示例一基础采样请求让我们先写一个最简单的采样示例。假设我们有一个 AI 助手它需要从本地日志文件中采样错误信息。python# simple_sampling.py# 演示 MCP Sampling 的基础用法from mcp import MCPClient, SamplingRequestimport asyncioasync def main(): # 创建 MCP 客户端 client MCPClient() # 建立连接 await client.connect(http://localhost:8000) # 创建采样请求 request SamplingRequest( prompt从最近的日志文件中找出前5条ERROR级别的日志记录, max_tokens200, # 限制返回长度 temperature0.1 # 低温度让结果更确定 ) print(发送采样请求...) # 发送请求并等待响应 response await client.sample(request) # 处理响应 if response.samples: print(f成功获取 {len(response.samples)} 条采样结果:) for idx, sample in enumerate(response.samples, 1): print(f{idx}. {sample.content}) else: print(未找到匹配的日志样本) # 关闭连接 await client.close()# 运行异步主函数if __name__ __main__: asyncio.run(main())这个示例展示了最基础的采样流程。SamplingRequest对象告诉服务器我们需要什么服务器处理后返回SamplingResponse包含实际的样本内容。max_tokens参数在这里特别重要它可以防止模型返回过长或过短的结果。## 示例二带人类协作的交互式采样实际应用中很多场景需要人类参与验证。比如AI 在分析医疗报告时遇到不确定的病例可以请求医生手动采样。下面这个示例演示了如何实现这种交互。python# collaborative_sampling.py# 展示 MCP Sampling 的人类协作机制from mcp import MCPClient, SamplingRequest, HumanSamplingHandlerimport asyncioclass MedicalSamplingHandler(HumanSamplingHandler): 自定义处理人类采样的回调 async def on_sampling_request(self, request: SamplingRequest): 当需要人类协助采样时触发 print(\n *50) print(【需要人类专家协助】) print(f请求说明: {request.prompt}) print(请根据以下选项输入响应:) print(1. 提供样本数据) print(2. 拒绝并提供理由) print(3. 请求更多信息) print(*50) # 模拟人类输入实际应用中可以是图形界面或API choice input(请输入选项 (1/2/3): ) if choice 1: sample_data input(请输入样本数据: ) return { status: accepted, samples: [{content: sample_data, metadata: {source: human}}] } elif choice 2: reason input(请输入拒绝理由: ) return { status: rejected, reason: reason } else: return { status: request_more, additional_info: input(请输入需要补充的信息: ) }async def main(): # 创建客户端并注册人类采样处理器 client MCPClient() client.set_human_sampling_handler(MedicalSamplingHandler()) try: await client.connect(ws://localhost:8001/mcp) print(已连接到 MCP 服务器支持人类协作) # 创建一个需要人类参与的采样请求 request SamplingRequest( prompt分析以下患者的CT影像判断是否存在异常阴影。 如果无法确定请请求人类专家协助。, max_tokens500, require_human_assistanceTrue # 开启人类协作模式 ) print(\n发起包含人类协作的采样请求...) response await client.sample(request) # 处理响应 if response.status accepted: print(f\n采样成功结果: {response.samples}) elif response.status rejected: print(f\n人类专家拒绝采样理由: {response.reason}) else: print(f\n需要补充信息: {response.additional_info}) except Exception as e: print(f发生错误: {e}) finally: await client.close()if __name__ __main__: asyncio.run(main())这个示例的关键在于HumanSamplingHandler类。它定义了当服务器需要人类协助时应该如何处理。通过设置require_human_assistanceTrue我们告诉 MCP 协议这个请求可能涉及需要人类判断的内容。在实际生产环境中这个处理器可以连接到 Slack 通知、任务队列或者图形界面。## 高级技巧采样参数调优在使用 MCP Sampling 时参数的调整直接影响采样质量1.temperature温度值越大如 1.0采样结果越多样化值越小如 0.1结果越确定。对于需要精确数据的场景建议使用低温度。2.max_tokens控制返回数据的长度。过小可能导致信息不全过大会浪费资源。3.stop_sequences可以设置停止标记让采样在遇到特定内容时自动终止。例如设置[ERROR, FATAL]可以让采样在发现错误时停止。4.require_human_assistance这个布尔标志决定是否允许人类介入。对于高风险场景如医疗、金融建议设为True。## 常见问题与调试-采样超时如果采样请求长时间没有响应检查max_tokens是否设置过大或者网络连接是否稳定。-采样结果为空可能是prompt描述不够具体尝试增加上下文信息。-人类协作失败确认HumanSamplingHandler是否正确注册以及回调函数是否正常处理输入。## 总结MCP Sampling 是一种强大而灵活的机制它让 AI 智能体能够智能地请求额外数据同时保持与人类专家的协作通道。通过本文的两个示例我们看到了从基础采样到人类协作的完整实现路径。在实际应用中你可以根据需求调整采样参数结合自定义处理器来适配各种业务场景。记住MCP Sampling 的核心价值在于「智能地知道何时需要更多信息」而不是盲目采样。这种机制让 AI 系统既高效又可靠是构建复杂智能应用的重要基石。随着 MCP 协议的不断发展未来我们可能会看到更多创新的采样模式比如多轮对话采样、基于元学习的自适应采样等。掌握好这个基础工具就能在 AI 应用开发中走得更远。