Patronus MCP Server

Patronus SDK 的一个MCP服务器实现，提供了一个标准化接口用于运行强大的LLM系统优化、评估和实验。

特性

• 使用API密钥和项目设置初始化Patronus
• 运行带有可配置评估器的单个评估
• 运行带有多个评估器的批量评估
• 运行带有数据集的实验

安装

1. 克隆仓库：

git clone https://github.com/yourusername/patronus-mcp-server.git
cd patronus-mcp-server

2. 创建并激活虚拟环境：

python -m venv .venv
source .venv/bin/activate  # 在Windows上：.venv\Scripts\activate

3. 安装主依赖和开发依赖：

uv pip install -e .
uv pip install -e ".[dev]"

使用方法

运行服务器

服务器可以通过两种方式提供的API密钥来运行：

1. 命令行参数：

python src/patronus_mcp/server.py --api-key your_api_key_here

2. 环境变量：

export PATRONUS_API_KEY=your_api_key_here
python src/patronus_mcp/server.py

交互式测试

测试脚本（tests/test_live.py）提供了以交互方式测试不同评估端点的方法。你可以通过几种方式运行它：

1. 命令行中的API密钥：

python -m tests.test_live src/patronus_mcp/server.py --api-key your_api_key_here

2. 环境中的API密钥：

export PATRONUS_API_KEY=your_api_key_here
python -m tests.test_live src/patronus_mcp/server.py

3. 不带API密钥（会提示输入）：

python -m tests.test_live src/patronus_mcp/server.py

测试脚本提供了三种测试选项：

1. 单个评估测试
2. 批量评估测试

每个测试都会以格式化的JSON输出显示结果。

API使用

初始化

from patronus_mcp.server import mcp, Request, InitRequest

request = Request(data=InitRequest(
    project_name="MyProject",
    api_key="your-api-key",
    app="my-app"
))
response = await mcp.call_tool("initialize", {"request": request.model_dump()})

单个评估

from patronus_mcp.server import Request, EvaluationRequest, RemoteEvaluatorConfig

request = Request(data=EvaluationRequest(
    evaluator=RemoteEvaluatorConfig(
        name="lynx",
        criteria="patronus:hallucination",
        explain_strategy="always"
    ),
    task_input="What is the capital of France?",
    task_output="Paris is the capital of France."
    task_context=["The capital of France is Paris."],
))
response = await mcp.call_tool("evaluate", {"request": request.model_dump()})

批量评估

from patronus_mcp.server import Request, BatchEvaluationRequest, RemoteEvaluatorConfig

request = Request(data=BatchEvaluationRequest(
    evaluators=[
        AsyncRemoteEvaluatorConfig(
            name="lynx",
            criteria="patronus:hallucination",
            explain_strategy="always"
        ),
        AsyncRemoteEvaluatorConfig(
            name="judge",
            criteria="patronus:is-concise",
            explain_strategy="always"
        )
    ],
    task_input="What is the capital of France?",
    task_output="Paris is the capital of France."
    task_context=["The capital of France is Paris."],
))
response = await mcp.call_tool("batch_evaluate", {"request": request.model_dump()})

运行实验

from patronus_mcp import Request, ExperimentRequest, RemoteEvaluatorConfig, CustomEvaluatorConfig

# 创建自定义评估函数
@evaluator()
def exact_match(expected: str, actual: str, case_sensitive: bool = False) -> bool:
    if not case_sensitive:
        return expected.lower() == actual.lower()
    return expected == actual

# 创建自定义适配器类
class ExactMatchAdapter(FuncEvaluatorAdapter):
    def __init__(self, case_sensitive: bool = False):
        super().__init__(exact_match)
        self.case_sensitive = case_sensitive

    def transform(self, row, task_result, parent, **kwargs):
        args = []
        evaluator_kwargs = {
            "expected": row.gold_answer,
            "actual": task_result.output if task_result else "",
            "case_sensitive": self.case_sensitive
        }
        return args, evaluator_kwargs

# 创建实验请求
request = Request(data=ExperimentRequest(
    project_name="my_project",
    experiment_name="my_experiment",
    dataset=[{
        "input": "What is 2+2?",
        "output": "4",
        "gold_answer": "4"
    }],
    evaluators=[
        # 远程评估器
        RemoteEvaluatorConfig(
            name="judge",
            criteria="patronus:is-concise"
        ),
        # 自定义评估器
        CustomEvaluatorConfig(
            adapter_class="my_module.ExactMatchAdapter",
            adapter_kwargs={"case_sensitive": False}
        )
    ]
))

# 运行实验
response = await mcp.call_tool("run_experiment", {"request": request.model_dump()})
response_data = json.loads(response[0].text)

# 实验异步运行，因此初始结果将是待定状态
assert response_data["status"] == "success"
assert "results" in response_data
assert isinstance(response_data["results"], str)  # 结果将是字符串（待定）

列出评估器信息

获取所有可用评估器及其相关标准的全面视图：

# 不需要请求体
response = await mcp.call_tool("list_evaluator_info", {})

# 响应结构：
{
    "status": "success",
    "result": {
        "evaluator_family_name": {
            "evaluator": {
                # 评估器配置和元数据
            },
            "criteria": [
                # 此评估器可用的标准列表
            ]
        }
    }
}

此端点将评估器及其相关标准的信息整合到一个组织良好的响应中。结果按评估器家族分组，每个家族包含其评估器配置和可用标准列表。

创建标准

在Patronus API中创建新的评估器标准。

{
    "request": {
        "data": {
            "name": "my-criteria",
            "evaluator_family": "Judge",
            "config": {
                "pass_criteria": "The MODEL_OUTPUT should contain all the details needed from RETRIEVED CONTEXT to answer USER INPUT.",
                "active_learning_enabled": false,
                "active_learning_negative_samples": null,
                "active_learning_positive_samples": null
            }
        }
    }
}

参数：

• name (str): 标准的唯一名称
• evaluator_family (str): 评估器家族（例如，“Judge”，“Answer Relevance”）
• config (dict): 标准的配置
  • pass_criteria (str): 必须满足的通过标准
  • active_learning_enabled (bool, 可选): 是否启用主动学习
  • active_learning_negative_samples (int, 可选): 主动学习的负样本数量
  • active_learning_positive_samples (int, 可选): 主动学习的正样本数量

返回：

{
    "status": "success",
    "result": {
        "name": "my-criteria",
        "evaluator_family": "Judge",
       
        "config": {
            "pass_criteria": "The MODEL_OUTPUT should contain all the details needed from RETRIEVED CONTEXT to answer USER INPUT.",
            "active_learning_enabled": False,
            "active_learning_negative_samples": null,
            "active_learning_positive_samples": null
        }
    }
}

自定义评估

使用装饰有@evaluator的自定义评估函数来评估任务输出。

{
    "request": {
        "data": {
            "task_input": "What is the capital of France?",
            "task_context": ["The capital of France is Paris."],
            "task_output": "Paris is the capital of France.",
            "evaluator_function": "is_concise",
            "evaluator_args": {
                "threshold": 0.7
            }
        }
    }
}

参数：

• task_input (str): 输入提示
• task_context (List[str], 可选): 评估上下文信息
• task_output (str): 要评估的输出
• evaluator_function (str): 要使用的评估函数名称（必须用@evaluator装饰）
• evaluator_args (Dict[str, Any], 可选): 评估函数的附加参数

评估函数可以返回：

• bool: 简单的通过/失败结果
• int或float: 数值分数（通过阈值是0.7）
• str: 文本输出
• EvaluationResult: 包含分数、通过状态、解释等的完整评估结果

返回：

{
    "status": "success",
    "result": {
        "score": 0.8,
        "pass_": true,
        "text_output": "Good match",
        "explanation": "Output matches context well",
        "metadata": {
            "context_length": 1
        },
        "tags": ["high_score"]
    }
}

示例评估函数：

from patronus import evaluator, EvaluationResult

@evaluator
def is_concise(output: str) -> bool:
    """简单的评估器，检查输出是否简洁"""
    return len(output.split()) < 10

@evaluator
def has_score(output: str, context: List[str]) -> EvaluationResult:
    """基于上下文返回分数的评估器"""
    return EvaluationResult(
        score=0.8,
        pass_=True,
        text_output="Good match",
        explanation="Output matches context well",
        metadata={"context_length": len(context)},
        tags=["high_score"]
    )

开发

项目结构

patronus-mcp-server/
├── src/
│   └── patronus_mcp/
│       ├── __init__.py
│       └── server.py
├── tests/
│   └── test_server.py
    └── test_live.py
├── pyproject.toml
└── README.md

添加新功能

1. 在server.py中定义新的请求模型：

   class NewFeatureRequest(BaseModel):
       # 定义你的请求字段
       field1: str
       field2: Optional[int] = None
   

2. 使用@mcp.tool()装饰器实现新的工具函数：

   @mcp.tool()
   def new_feature(request: Request[NewFeatureRequest]):
       # 实现你的功能逻辑
       return {"status": "success", "result": ...}
   

3. 添加相应的测试：

   • 在test_server.py中添加API测试：

     def test_new_feature():
         request = Request(data=NewFeatureRequest(
             field1="test",
             field2=123
         ))
         response = mcp.call_tool("new_feature", {"request": request.model_dump()})
         assert response["status"] == "success"
     

   • 在test_live.py中添加交互式测试：

     async def test_new_feature(self):
         request = Request(data=NewFeatureRequest(
             field1="test",
             field2=123
         ))
         result = await self.session.call_tool("new_feature", {"request": request.model_dump()})
         await self._handle_response(result, "New feature test")
     

   • 将新测试添加到main()中的测试选择菜单

4. 更新README，包括：

   • 功能描述部分的新功能描述
   • API使用部分的API使用示例
   • 任何新的配置选项或要求

运行测试

测试脚本使用模型上下文协议（MCP）客户端与服务器通信。它支持：

• 交互式测试选择
• JSON响应格式化
• 合理的资源清理
• 多种API密钥输入方法

你也可以运行标准测试套件：

pytest tests/

运行服务器

python -m src.patronus_mcp.server

许可证

该项目根据Apache许可证2.0发布 - 详情见LICENSE文件。

贡献

1. 分叉仓库
2. 创建功能分支
3. 提交更改
4. 推送到分支
5. 创建拉取请求