Patronus MCP 服务器

一个用于 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, optional): 是否启用主动学习
  • active_learning_negative_samples (int, optional): 主动学习的负样本数量
  • active_learning_positive_samples (int, optional): 主动学习的正样本数量

返回：

{
    "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], optional): 评估的上下文信息
• task_output (str): 要评估的输出
• evaluator_function (str): 要使用的评估器函数的名称（必须用 @evaluator 装饰）
• evaluator_args (Dict[str, Any], optional): 评估器函数的其他参数

评估器函数可以返回：

• 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:
    """Simple evaluator that checks if the output is concise"""
    return len(output.split()) < 10

@evaluator
def has_score(output: str, context: List[str]) -> EvaluationResult:
    """Evaluator that returns a score based on context"""
    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 License 2.0 获得许可 - 有关详细信息，请参阅 LICENSE 文件。

贡献

1. Fork 仓库
2. 创建一个功能分支
3. 提交您的更改
4. 推送到分支
5. 创建一个 Pull Request