[MCP][]Sampling示例

MCP Sampling 示例：让 AI 智能体与人类协作采样

引言：什么是 MCP Sampling？想象一下，你正在开发一个智能助手，它需要从海量数据中筛选出最有价值的信息。如果让它自己盲目采样，可能会遗漏重要内容；如果完全依赖人工，又效率低下。这时候，MCP（Model 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_tokens=200, # 限制返回长度 temperature=0.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_tokens=500, require_human_assistance=True # 开启人类协作模式 ) 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_assistance=True，我们告诉 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 应用开发中走得更远。