Token导航 LogoToken导航TokenDH.com
Langgraph Adaptive Tool Middleware logo
运维云端stdio官方级别未说明来源级核验

Langgraph Adaptive Tool Middleware

MCP Server

为LangGraph提供的生产级工具执行中间件,具备自动重试、熔断机制和错误恢复功能,适用于处理网络超时、API错误、速率限制和瞬时故障。

工具数

0

提示词数

0

GitHub Stars

0

资源数

0
Python云端部署Docker

安装说明

本站只整理中文说明和来源信息,不托管安装包,也不代用户安装。

作者 / 组织

dayal-arnav05

提供方

dayal-arnav05

最后核验

2026/5/17 20:22

快速接入

先看主来源和安装命令,再打开仓库或文档;下面只保留这个条目的关键接入事实。

命令预览

pip install -e .

详细介绍

LangGraph工具中间件

LangGraph的生产就绪工具执行中间件,具有自动重试、断路器和错误恢复功能。

![Python 3.8+](https://www.python.org/downloads/) ![License: MIT](https://opensource.org/licenses/MIT) ![Tests: 34K+](test_results/TEST_SUMMARY.md)

处理导致工具调用崩溃的实际生产故障:网络超时、API错误、速率限制和瞬时故障。

安装

# From source (for now)
pip install -e .

# Once published to PyPI
pip install langgraph-tool-middleware

安装.md 了解详细的安装说明。

快速开始

from src.production_middleware import ProductionToolMiddleware

# Initialize
middleware = ProductionToolMiddleware(
    enable_circuit_breaker=True,
    enable_error_feedback=True,
    max_retries=3
)

# Execute a tool with automatic retry
result = await middleware.execute(
    tool_call={"name": "get_weather", "arguments": {"location": "NYC"}},
    user_query="What's the weather in NYC?"
)

为什么存在

98%的工具调用失败是执行失败,而不是验证错误。

此中间件侧重于生产中实际发生的故障:

  • 网络超时(35%的故障)→ 使用指数回退重试
  • API错误(28%的失败)→ 智能重试逻辑
  • 速率限制(15%的故障)→ 断路器型式
  • 瞬态故障(20%的故障)→ 智能重试

经过验证的结果

在多种故障场景中进行34000多次测试:

场景基线中间件改进
低失败率(~17%)80%成功99.9%成功+20%
中等失败率(约35%)65%成功99.2%成功+35%
高失败率(~52%)50%成功98%成功+47%
极端失败(约70%)40%成功96%成功+57%

统计显著性: Z分数138.93(几乎可以肯定)

查看完整结果: test_results/TEST_SUMMARY.md

安装

pip install -r requirements.txt

在中设置OpenAI API密钥 .env:

OPENAI_API_KEY=your_key_here

LangGraph集成

插入式替换(3行!)

from src.production_middleware import ProductionToolMiddleware
from src.langgraph_integration import create_resilient_tool_node

# 1. Initialize middleware
middleware = ProductionToolMiddleware(max_retries=3)

# 2. Create your graph as normal
workflow.add_node("agent", call_model)

# 3. Use resilient tool node (ONE LINE!)
workflow.add_node("tools", create_resilient_tool_node(middleware))

# That's it! Your tools now have retry, circuit breakers, and recovery.

examples/langgraph_quickstart.py 举一个完整的例子。

独立使用

from src.production_middleware import ProductionToolMiddleware

# Initialize
middleware = ProductionToolMiddleware()

# Execute with automatic retry
result = await middleware.execute(
    tool_call={"name": "get_weather", "arguments": {"location": "London"}},
    user_query="What's the weather in London?"
)

if result['success']:
    print(f"✅ {result['result']}")
else:
    print(f"❌ {result['error']}")

建筑

核心组件

  1. 弹性执行者 (src/resilient_executor.py)

- 使用指数回退重试 - 超时处理 - 速率限制检测

  1. 断路器 (src/circuit_breaker.py)

- 防止级联故障 - 冷却后自动恢复 - 每个工具的运行状况跟踪

  1. 误差反馈 (src/error_feedback.py)

- 将真实错误反馈给LLM - 根据实际故障进行纠正 - 不仅仅是预测

  1. 可观测性 (src/observability.py)

- 每个工具的成功率 - 错误模式 - 延迟跟踪

  1. 主中间件 (src/production_middleware.py)

- 协调所有组件 - 清洁API - 生产准备就绪

测试

快速演示(查看实际操作)

# Show 20 baseline failures vs middleware recovery
python test/compare.py 20

这将运行测试,直到基线失败,然后显示中间件的功能。典型输出:

  • 基线: ❌ 失败(超时/速率限制/API错误)
  • 中间件: ✅ 尝试2-3次后成功(恢复率90%)

完整测试套件

# Comprehensive test (34,000+ tests)
python test/comprehensive_test.py

# Scale test (custom size)
echo "5000" | python test/test_at_scale.py

# Quick validation
python test/test_production_middleware.py

配置

middleware = ProductionToolMiddleware(
    # Retry settings
    max_retries=3,              # Max retry attempts
    retry_delay=1.0,            # Base delay between retries
    max_retry_delay=60.0,       # Max delay (exponential backoff)
    timeout=30.0,               # Max execution time
    
    # Circuit breaker
    enable_circuit_breaker=True,
    failure_threshold=5,        # Failures before circuit opens
    recovery_timeout=60.0,      # Time before retry
    half_open_max_calls=1,      # Test calls during recovery
    
    # Error feedback
    enable_error_feedback=True,
    model="gpt-4o-mini",       # LLM for corrections
    max_correction_attempts=2,  # Correction retries
    
    # Observability
    enable_observability=True
)

何时使用

高价值:

  • 网络状况不稳定
  • 具有速率限制的第三方API
  • 具有瞬态故障的微服务
  • 生产失败率>20%

中等价值:

  • 稳定的基础设施,偶尔会出现问题
  • 10-20%的故障率

低值:

  • 完善的基础设施(故障率\<5%)
  • 完全可靠的API(罕见)

项目结构

├── src/                    # Production middleware
│   ├── production_middleware.py
│   ├── resilient_executor.py
│   ├── circuit_breaker.py
│   ├── error_feedback.py
│   ├── observability.py
│   ├── tools.py           # Tool definitions
│   └── test_dataset.py    # Test cases
├── test/                   # Test scripts
│   ├── test_production_middleware.py
│   ├── test_at_scale.py
│   └── comprehensive_test.py
├── test_results/           # Test results
│   └── TEST_SUMMARY.md    # Full 34K+ test results
├── README.md              # This file
├── USAGE.md               # Detailed usage
└── requirements.txt

关键见解

验证中间件(HiTEC、ToolCritic)解决了约2%的问题(模式错误)。

该中间件解决了约98%的问题(执行失败)。

现代LLM已经很好地处理了模式验证。真正的问题是网络故障、API错误、速率限制和瞬态问题。这就是它解决的问题。

许可证

麻省理工学院

贡献

PR欢迎!重点领域:

  • 其他重试策略
  • 更复杂的断路器
  • 与LangGraph集成
  • MCP协议支持

目录标签

目录标签

Python云端部署Docker自动重试本地部署熔断机制错误恢复LangGraph集成生产级中间件

接入字段

传输方式(transport,传输协议)

stdio

鉴权方式(authType,认证方式)

none

工具数量(toolCount,工具数)

0

资源数量(resourceCount,资源数)

0

提示词数量(promptCount,提示词数)

0

权限和风险

stdionone部署方式未说明

接入前请确认传输方式、认证方式和部署位置,并根据实际工具能力限制访问范围。

安装前确认

不要直接授予不必要的文件、网络或账号权限;先核对安装命令和配置内容。

来源信息

继续浏览同类 MCP