1. MCP协议集成实战：让智能体连接外部世界

LangChain 1.0的create_agent API最令人兴奋的特性之一，就是能够无缝集成MCP协议服务。这就像给你的智能体装上了无数个外接设备插口，让它瞬间获得调用外部服务的能力。我在实际项目中用这个功能接过高德地图、天气API甚至企业内部CRM系统，效果比传统API调用方式流畅得多。

MCP协议本质上是个标准化接口，它解决了大模型与外部服务之间的"语言不通"问题。举个例子，当你想让智能体查询两地距离时，传统方式需要自己写地图API调用代码，而通过MCP协议，智能体能像调用本地函数一样直接使用远程服务。这背后的魔法在于MCP服务会提供标准的工具描述信息，就像函数的文档字符串，智能体看到这些说明就知道该怎么用了。

具体实现时，你需要先安装MCP适配器：

pip install langchain-mcp-adapters

然后配置MCP客户端连接参数。以高德地图服务为例，代码长这样：

from langchain_mcp_adapters.client import MultiServerMCPClient

mcp_client = MultiServerMCPClient({
    "amap-maps": {
        "command": "cmd",
        "args": ["/c", "npx", "-y", "@amap/amap-maps-mcp-server"],
        "env": {"AMAP_MAPS_API_KEY": "你的高德API密钥"},
        'transport': 'stdio'
    }
})

这里有个坑我踩过：MCP服务默认用stdio通信，如果服务端没正确启动会报连接错误。建议先用npx @amap/amap-maps-mcp-server手动测试服务能否正常运行。

获取工具列表的异步方法需要特别注意：

async def get_tools():
    tools = await mcp_client.get_tools()
    print(f"加载工具: {[t.name for t in tools]}")
    
asyncio.run(get_tools())

看到控制台输出工具列表后，就可以用create_agent构建智能体了。我习惯加个系统提示明确角色：

agent = create_agent(
    model,
    tools=tools,
    system_prompt="你是个专业地图助手，用简洁语言回答导航问题"
)

实测发现，当工具超过5个时，智能体开始会出现选择困难。这时可以用tool_choice参数指定默认工具，或者在中间件里做工具过滤，这个后面会详细讲。

2. 中间件深度优化：给智能体装上"调节器"

如果说MCP扩展了智能体的能力边界，那么中间件就是控制这些能力的"调节器"。在最近的一个客服项目中，我们通过中间件实现了请求日志、敏感词过滤和耗时统计，把错误率降低了60%。create_agent的中间件机制提供了6个关键介入点：

• before_agent：适合做权限校验
• before_model：可以修改提示词
• wrap_model_call：能拦截模型请求
• wrap_tool_call：可监控工具执行
• after_model：用于结果校验
• after_agent：适合资源清理

预置的HumanInTheLoopMiddleware特别实用。有次我们智能体要执行数据库写入操作，就是靠它添加人工确认环节，避免了灾难性错误。配置示例：

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model,
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={"dangerous_tool": {"require_approval": True}}
    )]
)

当触发危险操作时，流程会暂停并返回__interrupt__标志。继续执行需要发送确认指令：

if "__interrupt__" in result:
    result = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config
    )

自定义中间件更有趣。我做过一个根据用户等级切换模型的中间件，VIP用户用GPT-4，普通用户用Claude：

class ModelSwitchMiddleware(AgentMiddleware):
    def wrap_model_call(self, request, handler):
        if request.context.get('vip'):
            request.model = gpt4_model
        else:
            request.model = claude_model
        return handler(request)

中间件执行顺序很重要！曾经因为把日志中间件放最后，漏记了关键错误。建议按功能排序：安全类→业务类→监控类。

3. 性能优化实战：从理论到实践

MCP服务虽然方便，但性能问题不容忽视。在压力测试中，我们发现三个瓶颈点：工具加载延迟、网络通信开销和模型推理耗时。通过以下优化手段，最终QPS从15提升到了42：

首先是工具懒加载。默认情况下get_tools()会加载全部工具，改用按需加载：

async def get_tool(name):
    return await mcp_client.get_tool(name)

其次是连接池优化。MCP客户端默认单连接，高并发时会排队：

mcp_client = MultiServerMCPClient(
    config,
    pool_size=5  # 根据服务端承受能力调整
)

缓存是另一个利器。我们对工具描述信息做本地缓存，TTL设10分钟：

from datetime import timedelta
from langchain.cache import InMemoryCache

InMemoryCache(ttl=timedelta(minutes=10))

中间件层面的优化更精彩。我们开发了智能节流中间件，当检测到超时自动降级：

class ThrottleMiddleware(AgentMiddleware):
    def wrap_model_call(self, request, handler):
        start = time.time()
        try:
            return handler(request)
        except TimeoutError:
            request.model = backup_model  # 切换到轻量模型
            return handler(request)
        finally:
            record_latency(time.time() - start)

监控方面，推荐在after_agent钩子里埋点：

class MetricsMiddleware(AgentMiddleware):
    def after_agent(self, context):
        statsd.gauge('agent.duration', context['duration'])
        if context.get('error'):
            statsd.increment('agent.errors')

这些优化手段需要根据实际场景组合使用。我们的经验是：先监控定位瓶颈，再针对性优化，最后用A/B测试验证效果。

4. 安全防护体系构建

去年某次安全审计中，我们发现智能体存在三大风险：敏感信息泄露、越权工具调用和提示词注入。通过中间件构筑了三道防线：

第一道是输入过滤，用PIIMiddleware自动脱敏：

from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model,
    middleware=[PIIMiddleware(
        patterns=[r'\d{4}-\d{8}', r'1[3-9]\d{9}']  # 身份证和手机号
    )]
)

第二道是权限控制，我们扩展了HumanInTheLoopMiddleware：

class AuthMiddleware(AgentMiddleware):
    def before_tool_call(self, tool, input):
        if tool.requires_auth and not context.get('auth'):
            raise PermissionError("请先登录")

第三道是输出过滤，防止返回危险内容：

class SafetyMiddleware(AgentMiddleware):
    def after_model(self, response):
        if contains_harmful_content(response.text):
            response.text = "内容违反安全策略"
        return response

日志审计也很关键。我们开发了可追溯中间件，为每个请求生成唯一ID：

class TraceMiddleware(AgentMiddleware):
    def before_agent(self, context):
        context['request_id'] = uuid.uuid4()
        log_request(context)

最近我们还加入了频控中间件，防止API滥用：

class RateLimitMiddleware(AgentMiddleware):
    def __init__(self):
        self.counter = defaultdict(int)
        
    def before_agent(self, context):
        ip = context['client_ip']
        if self.counter[ip] > 100:
            raise RateLimitError("请求过于频繁")
        self.counter[ip] += 1

安全是个持续过程，我们每月会做红蓝对抗演练，不断发现和修补漏洞。建议至少每季度做一次全面安全评估。