Patronus MCP服务器
Patronus SDK的MCP服务器实现,为运行强大的LLM系统优化、评估和实验提供了标准化的接口。
特性
- 使用API密钥和项目设置初始化Patronus
- 使用可配置的评估器运行单次评估
- 使用多个评估者运行批量评估
- 使用数据集进行实验
安装
- 克隆存储库:
git clone https://github.com/yourusername/patronus-mcp-server.git
cd patronus-mcp-server- 创建并激活虚拟环境:
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate- 安装主依赖项和开发依赖项:
uv pip install -e .
uv pip install -e ".[dev]"用法
运行服务器
服务器可以使用API密钥运行,该密钥以两种方式提供:
- 命令行参数:
python src/patronus_mcp/server.py --api-key your_api_key_here- 环境变量:
export PATRONUS_API_KEY=your_api_key_here
python src/patronus_mcp/server.py交互式测试
测试脚本(tests/test_live.py)提供了一种交互式方法来测试不同的评估端点。您可以通过多种方式运行它:
- 使用命令行中的API键:
python -m tests.test_live src/patronus_mcp/server.py --api-key your_api_key_here- 使用API密钥在环境中:
export PATRONUS_API_KEY=your_api_key_here
python -m tests.test_live src/patronus_mcp/server.py- 没有API密钥(将提示):
python -m tests.test_live src/patronus_mcp/server.py测试脚本提供了三个测试选项:
- 单项评价试验
- 批量评估测试
每个测试都将以格式良好的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
# Create a custom evaluator function
@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
# Create a custom adapter class
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
# Create experiment request
request = Request(data=ExperimentRequest(
project_name="my_project",
experiment_name="my_experiment",
dataset=[{
"input": "What is 2+2?",
"output": "4",
"gold_answer": "4"
}],
evaluators=[
# Remote evaluator
RemoteEvaluatorConfig(
name="judge",
criteria="patronus:is-concise"
),
# Custom evaluator
CustomEvaluatorConfig(
adapter_class="my_module.ExactMatchAdapter",
adapter_kwargs={"case_sensitive": False}
)
]
))
# Run the experiment
response = await mcp.call_tool("run_experiment", {"request": request.model_dump()})
response_data = json.loads(response[0].text)
# The experiment runs asynchronously, so results will be pending initially
assert response_data["status"] == "success"
assert "results" in response_data
assert isinstance(response_data["results"], str) # Results will be a string (pending)列出评估者信息
全面了解所有可用的评估人员及其相关标准:
# No request body needed
response = await mcp.call_tool("list_evaluator_info", {})
# Response structure:
{
"status": "success",
"result": {
"evaluator_family_name": {
"evaluator": {
# evaluator configuration and metadata
},
"criteria": [
# list of available criteria for this evaluator
]
}
}
}此端点将有关评估者及其相关标准的信息组合成一个有组织的响应。结果按评估器族分组,每个族包含其评估器配置和可用条件列表。
创建标准
在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):评估者的家庭(例如,“判断”、“答案相关性”)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:
"""Simple evaluator that checks if the output is concise"""
return len(output.split()) 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添加新功能
- 在中定义新的请求模型
server.py:
class NewFeatureRequest(BaseModel):
# Define your request fields here
field1: str
field2: Optional[int] = None- 使用以下工具实现新的工具功能
@mcp.tool()装饰师:
@mcp.tool()
def new_feature(request: Request[NewFeatureRequest]):
# Implement your feature logic here
return {"status": "success", "result": ...}- 添加相应的测试:
- 在中添加API测试 test_server.py:
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()
- 用以下内容更新README:
- “功能”部分中的新功能描述 - API用法部分中的API用法示例 - 任何新的配置选项或要求
运行测试
测试脚本使用模型上下文协议(MCP)客户端与服务器通信。它支持:
- 交互式测试选择
- JSON响应格式
- 适当的资源清理
- 多个API键输入法
您还可以运行标准测试套件:
pytest tests/运行服务器
python -m src.patronus_mcp.server许可证
此项目根据Apache许可证2.0获得许可-请参阅 许可证 文件以获取详细信息。
贡献
- 分叉存储库
- 创建要素分支
- 提交您的更改
- 推到分支
- 创建拉取请求
