TRM Python MCP服务器
A. TRM启发(测试时递归存储器)MCP服务器 用于Python数据分析项目中的递归代码细化。
此服务器实现了专门为Python和数据分析工作流设计的递归改进循环:
- LLM扮演着 优化器 建议代码更改
- 此MCP服务器充当 评论家/评价者 具有状态跟踪功能
- 评估包括: 数据质量检查、测试、lint和性能指标
- 使用EMA(指数移动平均线)跟踪分数
- 智能停止策略决定何时停止
✨ 特性
- 多信号评估:数据质量、测试、lint和性能基准测试
- 加权评分:不同评估信号的可配置权重
- EMA跟踪:跨迭代平滑跟踪分数
- 智能停机:当测试通过+分数阈值、无改善或最大步数时停止
- 灵活的候选人提交:支持多种模式(文件、补丁、差异)
- 安全执行:命令在具有可配置超时的隔离目录中运行
- Python专用:回溯解析、pytest/unittest集成、Python linting
- 数据分析重点:Pandas/NumPy验证、数据质量检查
🐍 数据分析重点
此服务器是专门为Python数据分析项目设计的:
评估信号
- 数据质量 (默认权重:0.3)
- 架构验证 - 缺少值检查 - 数据类型一致性 - Pandas DataFrame验证 - 自定义数据质量脚本
- 测试 (默认权重:0.4)
- pytest集成 - 单元测试支持 - JSON测试报告器 - 测试覆盖率指标
- 棉绒 (默认权重:0.1)
- flake8样式检查 - pylint代码质量 - mypy类型检查 - PEP 8合规性
- 演出 (默认权重:0.2)
- 运行时指标 - 内存配置文件 - 数据处理基准 - 标准化绩效评分
📦 安装
先决条件
- Python 3.10或更高版本
- pip用于包管理
从源代码安装
git clone https://github.com/andreahaku/code_trm_python_mcp.git
cd code_trm_python_mcp
pip install -e .开发安装
pip install -e ".[dev]"🚀 用法
作为MCP服务器
添加到MCP客户端配置中:
{
"mcpServers": {
"code-trm-python": {
"command": "python",
"args": ["-m", "src.server"]
}
}
}或者使用已安装的脚本:
{
"mcpServers": {
"code-trm-python": {
"command": "code-trm-python-mcp"
}
}
}💬 示例用法
在LLM客户端中配置为MCP服务器后,您可以使用自然语言提示:
启动TRM会话
提示: *“在/Users/me/projects/data pipeline上启动我的数据分析项目的TRM会话。使用pytest进行测试,使用flake8和mypy进行linting,并使用以下命令运行数据验证 python scripts/validate_schemas.py.将最大步数设置为10,当分数达到0.95时停止。”*
这将调用 trm.start 工具用于:
- 初始化指定项目上的会话
- 配置数据质量、测试和lint的评估命令
- 设置停止策略(最大步数=10,阈值=0.95)
- 返回会话ID以进行后续操作
迭代改进
提示: *“提交一个修复src/predicting.py中缺失值处理的候选项。问题在第45-50行,我们需要为'age'列添加fillna()。”*
这将:
- 将建议的更改应用于文件
- 运行完整的评估管道(数据质量、测试、lint、perf)
- 计算加权分数
- 返回通过/失败的反馈
- 建议是继续还是停止
提示: *“TRM会话的当前状态是什么?告诉我分数以及我们是否应该停止迭代。”*
用途 trm.state 和 trm.halt 工具用于:
- 显示当前步骤、EMA得分、最佳得分
- 显示停止的决定和原因
- 提供连续改进信息
使用文件
提示: *“阅读src/pipeline.py和data/schemas.py的内容,这样我就可以看到需要修复的地方。”*
用途 trm.read 工具用于:
- 从存储库中获取文件内容
- 显示元数据(行数、大小、上次修改时间)
- 为做出明智的改变提供背景
提示: *“显示发生错误的src/analysis.py的第100-120行。”*
用途 trm.lines 工具用于:
- 读取带有行号的特定行范围
- 围绕错误位置提供重点上下文
- 与读取整个文件相比,保存令牌
获取帮助
提示: *“测试因NameError而失败。你能建议一个修复方法吗?”*
用途 trm.fix 工具用于:
- 分析Python错误回溯
- 识别具体错误(NameError:未定义“pandas”)
- 生成修复候选(添加
import pandas as pd) - 为准备申请的候选人提供理由
提示: *“根据评估结果,我应该关注哪些改进?”*
用途 trm.suggest 工具用于:
- 分析评估反馈
- 确定问题的优先级(关键→ high → 中等→ low)
- 生成可操作的建议
- 考虑代码质量、测试覆盖率、性能
代码审查
提示: *“查看PRhttps://github.com/user/repo/pull/123并检查数据验证问题、缺失测试和类型安全问题。"*
用途 trm.review 工具用于:
- 从GitHub获取PR差异
- 分析代码更改
- 检查常见问题(缺少类型提示、没有错误处理等)
- 提供详细的意见和建议
- 进行总体评估(批准/需要更改/评论)
检查点和撤消
提示: *将当前状态另存为名为“重构前”的检查点*
用途 trm.save 用于保存会话状态以供以后还原的工具。
提示: *“最后的更改使情况变得更糟。撤消它并恢复到以前的状态。”*
用途 trm.undo 工具用于:
- 将文件还原到以前的状态
- 回滚步计数器
- 重新计算分数和EMA
- 显示已恢复的内容
示例完成会话
\*提示: “让我们改进我的pandas项目中的数据验证。使用以下设置在/path/to/project上启动TRM会话:
- 数据验证:python脚本/check_schemas.py
- 测试:pytest--json测试/
- 棉绒:flake8型钢圈/&&mypy型钢圈/
- 性能:python脚本/benchmark_pipeline.py
- 12步后或得分达到0.97时停止
答复: ✅ 会话以ID abc-123开始。。。
提示: *“阅读src/vvalidators.py,这样我就可以看到需要改进的地方。”*
答复: \[显示250行的文件内容\]
提示: *“提交一个为‘事务’DataFrame添加架构验证的候选项。它应该检查所需的列是否存在并且具有正确的数据类型。”*
答复:
- 第1/12步
- 得分:0.65
- 测试:45/50通过(5失败)
- 数据质量:❌ 失败(检测到架构不匹配)
- 反馈:添加对“金额”列数据类型的验证
提示: *“修复基于该反馈的架构验证。”*
答复:
- 第2/12步
- 得分:0.82
- 测试:48/50通过(2失败)
- 数据质量:✅ 通过
- 反馈:边缘情况下的剩余测试失败
提示: *“继续迭代,直到测试通过,否则我们应该停止。”*
\[系统继续迭代LLM提案…\]
答复:
- 第5/12步
- 得分:0.96
- 测试:50/50通过✅
- 数据质量:✅ 通过
- 应该停止:是的,成功!通过测试且得分>=阈值
🛠️ 可用工具(共16个)
核心工具
trm.start -开始会话
在本地存储库上初始化TRM会话。
参数:
repo(必填):项目的绝对路径dataQual:数据质量验证命令test:测试命令(例如“pytest--json”)lint:Lint命令(例如,“flake8 src/”)bench:性能基准命令timeout:每个命令超时(默认值:120秒)weights:得分权重(数据质量:0.3,测试:0.4,皮棉:0.1,性能:0.2)halt:暂停策略(最大值、阈值、耐心、最小值)ema:EMA平滑系数(默认值:0.9)notes:可选的初始推理说明
退货: sessionId, message,配置详细信息
例子:
session = await trm.start({
"repo": "/path/to/project",
"dataQual": "python scripts/validate_data.py",
"test": "pytest --json tests/",
"lint": "flake8 src/ && mypy src/",
"bench": "python scripts/benchmark.py",
"halt": {
"max": 12,
"threshold": 0.95,
"patience": 3
}
})trm.submit -提交候选人
应用候选人变更,运行评估,返回反馈。
参数:
sid(必填):会话IDcandidate(必填):其中一种模式:
- 文件:完整的文件内容 - 补丁:统一的差异格式 - 差异: 每个文件的差异
reason:法学硕士推理笔记
退货: step, score, emaScore, bestScore, tests, okDataQuality, okLint, shouldHalt, reasons, feedback
例子:
result = await trm.submit({
"sid": session_id,
"candidate": {
"mode": "diff",
"changes": [{
"path": "src/analysis.py",
"diff": "--- a/src/analysis.py\\n+++ b/src/analysis.py\\n..."
}]
},
"reason": "Fix data validation logic"
})trm.read -获取文件内容
使用元数据读取当前文件状态。
参数:
sid(必填):会话IDpaths(必填):文件路径列表
退货: 带元数据的文件内容(行数、大小字节、lastModified)
trm.state -获取州
返回当前会话状态快照。
退货: sessionId, step, emaScore, bestScore, noImproveStreak, last, zNotes
trm.halt -应该停止
检查最新评估中的停止决定。
退货: shouldHalt, reasons
trm.end -结束会话
清理并结束会议。
退货: ok, message
增强工具
trm.validate -验证候选人
在应用更改之前进行详细预览的模拟运行验证。
参数: sid, candidate
退货: valid, errors, warnings, preview
trm.suggest -获取建议
根据评估结果获取基于人工智能的改进建议。
退货: 按优先级排序的热门建议
检查点工具
trm.save -保存检查点
将当前会话状态另存为检查点。
参数: sid, desc (可选描述)
trm.restore -还原检查点
从已保存的检查点还原会话。
参数: sid, cid (检查点ID)
trm.list -列出检查点
列出会话的所有已保存检查点。
参数: sid
高级工具
trm.undo -撤消上一个候选人
快速撤消并完全恢复状态。
退货: message, currentStep, score, emaScore, filesRestored
trm.lines -获取文件行
从带有行号的文件中读取特定的行范围。
参数: sid, file, start (行号), end (行号)
退货: 带格式化行号的行,总行数
trm.fix -建议修复
基于Python错误分析的人工智能修复候选生成。
支持的错误: NameError、ImportError、TypeError、AttributeError、SyntaxError
退货: 一系列建议 priority, issue, candidateToFix, rationale
trm.reset -重置为基线
将存储库重置为初始git提交状态。
trm.review -公关评论
对来自GitHub URL或直接差异的拉取请求进行详细的代码审查。
参数:
url:GitHub PR URLdiff:直接统一差异内容files:包含内容的文件数组focus:筛选评论类别的可选数组
重点类别:
type-safety:检测缺失的类型提示logging:标记打印语句todos:识别TODO/FIXME评论code-quality:魔术数字,长线error-handling:缺少尝试捕捉testing:建议添加测试data-validation:缺少数据检查
📊 分数计算
分数是\[0,1\]中的加权平均值:
score = (w.dataQual * sDataQual + w.test * sTests + w.lint * sLint + w.perf * sPerf) / sumWeights
where:
sDataQual = 1 if data quality checks pass, 0 otherwise
sTests = passed / total (0 if tests fail to parse)
sLint = 1 if lint succeeds, 0 otherwise
sPerf = normalized performance score (best/current, lower runtime is better)🛑 停机条件
迭代在以下情况下停止:
- 成功:
step >= minSteps所有测试均已通过score >= passThreshold - 高原:没有改善
patience连续步骤 - 限制:已到达
maxSteps
🎯 推荐工作流程
1.开始会话
session = await trm.start({
"repo": "/path/to/data-analysis-project",
"dataQual": "python scripts/validate_schemas.py",
"test": "pytest --json tests/",
"lint": "flake8 src/ && mypy src/",
"bench": "python scripts/benchmark_pipeline.py",
"halt": {"max": 12, "threshold": 0.97, "patience": 3}
})2.迭代改进循环
关键原则:
- 保留补丁 小而专注 (一次一个问题)
- 使用
reason跨步骤维护上下文 - 相信分数/反馈信号作为指导
图案:
- 先获取文件内容
- 提交候选人并说明理由
- 如果失败:使用
trm.fix或调整方法 - 重复,直到
shouldHalt=true
3.迭代示例
# 1. Get current files
files = await trm.read({"sid": session_id, "paths": ["src/pipeline.py"]})
# 2. Submit improvement
result = await trm.submit({
"sid": session_id,
"candidate": {
"mode": "diff",
"changes": [{
"path": "src/pipeline.py",
"diff": "..."
}]
},
"reason": "Add data validation for missing values"
})
# 3. Check if should continue
if result["shouldHalt"]:
print(f"Done! Final score: {result['score']}")
else:
print(f"Continue - Step {result['step']}, Score: {result['score']}")🐍 Python特定功能
解析时出错
自动解析Python回溯:
- 名称错误→ 建议导入或变量定义
- 导入错误→ 建议
pip install命令 - 类型错误→ 识别参数不匹配
- 属性错误→ 检测不正确的属性访问
- 语法错误→ 突出语法问题
测试框架支持
- pytest:JSON输出解析、夹具检测
- 单元测试:标准输出解析、测试发现
- 覆盖率指标集成
Linting集成
- 薄片8:样式检查
- 皮林:代码质量分析
- 米皮:类型提示验证
数据分析验证
- Pandas DataFrame模式验证
- NumPy数组形状/dtype检查
- 缺失值检测
- 数据类型一致性验证
💡 数据分析项目提示
- 设置数据质量检查 验证:
- DataFrame架构与预期结构匹配 - 无意外缺失值 - 数据类型正确 - 值范围有效
- 使用有意义的绩效基准:
- 时间关键型数据处理管道 - 测量大型数据集的内存使用情况 - 跨迭代跟踪改进/回归
- 配置适当的权重:
- 更高 test 算法正确性的权重 - 更高 dataQual 数据管道可靠性权重 - 更高 perf 优化任务的权重
- 利用pytest标记 对于不同的测试类别:
- @pytest.mark.unit 用于快速单元测试 - @pytest.mark.integration 用于数据管道测试 - @pytest.mark.slow 用于长时间运行的测试
🏗️ 建筑
┌─────────────────────────────────────────────────────────────┐
│ LLM Client │
│ (Claude Code / Cursor / Codex CLI) │
│ │
│ • Proposes code changes (optimizer role) │
│ • Submits candidates via MCP tools │
│ • Interprets feedback and iterates │
└────────────────────┬────────────────────────────────────────┘
│ MCP Protocol
▼
┌─────────────────────────────────────────────────────────────┐
│ TRM Python MCP Server │
│ │
│ Session State: │
│ • Current score, EMA, best score │
│ • Test results, data quality status │
│ • Improvement streak tracking │
│ • History of evaluations │
│ • Candidate snapshots (for undo) │
│ │
│ Evaluation Pipeline: │
│ 1. Apply candidate changes │
│ 2. Run: data_quality → test → lint → perf │
│ 3. Parse outputs, extract signals │
│ 4. Compute weighted score │
│ 5. Update EMA and improvement tracking │
│ 6. Check halting policy │
│ 7. Return structured feedback │
└────────────────────┬────────────────────────────────────────┘
│ Shell Commands
▼
┌─────────────────────────────────────────────────────────────┐
│ Target Python Project │
│ │
│ • Source code files (.py, .ipynb) │
│ • Data validation scripts │
│ • Test framework (pytest, unittest) │
│ • Linters (flake8, pylint, mypy) │
│ • Performance benchmark scripts │
│ • Data analysis pipelines (pandas, numpy) │
└─────────────────────────────────────────────────────────────┘📝 基于
这一实施的灵感来自 测试时间递归存储器(TRM) 方法来自:
“递归反思:教授语言模型代理如何自我提升” (arXiv:2510.4871v1)
Python数据分析的关键调整:
- TRM的递归精化→ 使用LLM提案进行迭代代码改进
- 潜在推理(z)→ 迭代之间传递的基本原理/注释
- ACT停止→ 基于分数+改进的可配置停止策略
- 深度监管→ 数据质量/测试/lint/perf信号作为无需培训的反馈
- Python焦点→ 回溯解析、pytest集成、数据验证
📄 许可证
麻省理工学院
👨💻 作者
安德烈·萨尔瓦托(@andreahaku)与克劳德(Anthropic)
🔗 相关项目
- llm存储器mcp -LLM工具的持久内存
- 代码分析上下文python-mcp -Python项目的代码分析
- 代码trm-mcp -Types/JavaScript版本
🤝 贡献
欢迎投稿!请随时提交拉取请求。
______________________________________________________________________
状态: 🚧 贝塔 -核心功能已实现,高级功能正在开发中
Python版本: 3.10+
聚焦:数据分析和科学计算(Pandas、NumPy、Scikit-learn)
