代理scGPT
一个暴露的Pydantic AI代理 scGPT 通过模型上下文协议（MCP）将单细胞基础模型功能作为工具。
项目概述
该项目通过以下方式弥合了对话式人工智能代理和单细胞基因组学分析之间的差距：
- 将scGPT函数作为MCP工具公开 通过 FastMCP
- 创建Pydantic AI代理 可以编排这些工具
- 使用Temporal增加可靠性 用于重试逻辑和工作流编排
建筑
┌─────────────────────────────────────────────────────────────────┐
│ Pydantic AI Agent │
│ (Orchestration Layer) │
└─────────────────────────┬───────────────────────────────────────┘
│ MCP Protocol
▼
┌─────────────────────────────────────────────────────────────────┐
│ FastMCP Server │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ Cell Type │ │ Batch │ │ Gene Network │ │
│ │ Annotation │ │ Integration │ │ Inference │ │
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
└─────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Temporal Workflows │
│ (Retry Logic & Orchestration) │
└─────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ scGPT Model │
│ (GPU-accelerated inference) │
└─────────────────────────────────────────────────────────────────┘计划中的MCP工具（第一阶段）
工具1： annotate_cell_types
使用scGPT包埋基于基因表达谱对单细胞进行注释。
@mcp.tool
async def annotate_cell_types(
expression_data: str, # Path to h5ad file or CSV
reference_dataset: str = "cellxgene",
batch_size: int = 64
) -> dict:
"""
Annotate cell types from single-cell RNA-seq data.
Returns predicted cell types with confidence scores.
"""工具2： integrate_batches
整合多个scRNA-seq数据集，同时校正批处理效应。
@mcp.tool
async def integrate_batches(
dataset_paths: list[str], # Paths to multiple h5ad files
batch_key: str = "batch",
n_hvg: int = 2000
) -> dict:
"""
Integrate multiple single-cell datasets with batch correction.
Returns path to integrated dataset and integration metrics.
"""工具3： get_gene_embeddings
提取基因嵌入物用于下游分析或基因网络推理。
@mcp.tool
async def get_gene_embeddings(
gene_list: list[str],
model_checkpoint: str = "scGPT_human"
) -> dict:
"""
Get scGPT embeddings for a list of genes.
Returns gene embeddings that can be used for similarity analysis.
"""开发阶段
第一阶段：基础设置
- \[x\] 通过以下方式设置项目结构
uv - \[x\] 安装核心依赖项（scGPT、FastMCP、Pydantic AI）
- \[x\] 验证GPU/CUDA兼容性
- \[x\] 下载scGPT预训练检查点
第2阶段：MCP服务器实施
- \[x\] 创建FastMCP服务器骨架
- \[x\] 实施
annotate_cell_types工具 - \[x\] 实施
integrate_batches工具 - \[x\] 实施
get_gene_embeddings工具 - \[x\] 使用Pydantic模型添加输入验证
- \[x\] 使用编写工具测试 糖浆般的 快照测试
第三阶段：时间整合
- \[\]设置临时工
- \[\]将scGPT操作包装为时态活动
- \[\]为GPU操作添加重试策略
- \[\]实施多步骤分析的工作流程
第四阶段：Pydantic AI代理
- \[\]使用MCP服务器连接创建代理
- \[\]定义生物学领域的系统提示
- \[\]添加对话记忆/上下文
- \[\]实施示例分析工作流
第5阶段：测试和文件编制
- \[\]使用样本数据进行集成测试
- \[\]性能基准
- \[\]API文件
- \[\]使用示例和教程
安装
先决条件
- Python 3.11+
- 支持CUDA的GPU（推荐）
- 紫外线 包管理器
设置
# Clone the repository
git clone
cd agentic_scgpt
# Create virtual environment and install dependencies
uv sync
# Download scGPT checkpoints
python -m scripts.download_checkpoints
# Start Temporal (if using retry logic)
temporal server start-dev
# Run the MCP server
uv run python -m src.server项目结构
agentic_scgpt/
├── src/
│ ├── __init__.py
│ ├── server.py # FastMCP server with scGPT tools
│ ├── agent.py # Pydantic AI agent
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── annotate.py # Cell type annotation tool
│ │ ├── integrate.py # Batch integration tool
│ │ └── embeddings.py # Gene embeddings tool
│ ├── workflows/
│ │ ├── __init__.py
│ │ └── activities.py # Temporal activities
│ └── models/
│ ├── __init__.py
│ └── schemas.py # Pydantic schemas for tool I/O
├── scripts/
│ ├── check_gpu.py # GPU diagnostics
│ └── download_checkpoints.py
├── tests/
├── checkpoints/ # scGPT model weights
├── pyproject.toml
└── README.md依赖项
| 包装 | 用途 |
|---|---|
scgpt | 单细胞基础模型 |
fastmcp | MCP服务器框架 |
pydantic-ai | AI代理框架 |
temporalio | 工作流编排和重试 |
torch | 深度学习后端 |
scanpy | 单细胞分析实用程序 |
anndata | scRNA-seq的数据结构 |
示例用法
运行MCP服务器
# src/server.py
from fastmcp import FastMCP
mcp = FastMCP("scGPT Tools")
@mcp.tool
async def annotate_cell_types(expression_data: str) -> dict:
"""Annotate cell types from scRNA-seq data."""
# Implementation here
pass
if __name__ == "__main__":
mcp.run(transport="streamable-http", port=8000)连接Pydantic AI代理
# src/agent.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
server = MCPServerStreamableHTTP("http://localhost:8000/mcp")
agent = Agent(
"anthropic:claude-sonnet-4-20250514",
toolsets=[server],
system_prompt="""You are a single-cell genomics expert assistant.
You help researchers analyze scRNA-seq data using scGPT tools.
Always explain your analysis steps and interpret results."""
)
async def main():
async with agent:
result = await agent.run(
"Annotate the cell types in my dataset at data/pbmc.h5ad"
)
print(result.output)使用临时检索
# src/workflows/activities.py
from temporalio import activity
from datetime import timedelta
@activity.defn
async def annotate_cells_activity(data_path: str) -> dict:
"""Activity with automatic retry on GPU OOM errors."""
# scGPT annotation logic
pass
# In workflow
await workflow.execute_activity(
annotate_cells_activity,
"data/sample.h5ad",
start_to_close_timeout=timedelta(minutes=10),
retry_policy=RetryPolicy(
maximum_attempts=3,
initial_interval=timedelta(seconds=5),
non_retryable_error_types=["InvalidDataError"]
)
)GPU要求
scGPT需要一个支持CUDA的GPU来进行高效推理。运行GPU检查脚本：
uv run python scripts/check_gpu.py推荐规格：
- 配备8GB+VRAM的NVIDIA GPU
- 第11.8节+
- cuDNN 8.6+
资源
许可证
麻省理工学院