mirror of
https://github.com/Mesh-LLM/mesh-llm.git
synced 2026-08-08 22:23:19 -04:00
Add SkippyBench external eval pack (#865)
* Add SkippyBench external eval pack * Support structured eval requests in skippy OpenAI serving * Update skippy smoke structured output expectation * Guard external eval harness execution * Strengthen Docker eval preflight * Preflight external eval runs * Add external eval harness orchestration
This commit is contained in:
James Dumay
GitHub
parent
c39cb3b06c
commit
4f25060ca0
24 changed files with 3512 additions and 150 deletions
|
|
@ -9,6 +9,18 @@ metadata:
|
|||
|
||||
Use this skill for performance, orchestration, and report-oriented checks.
|
||||
Use `skippy-correctness` when the question is pass/fail exactness.
|
||||
All reportable benchmark runs need metrics-server. `run`, `focused-runtime`,
|
||||
and `local-single` start a collector by default; endpoint-driving commands such
|
||||
as `chat-corpus` and `eval run` require `--metrics-http` to point at an
|
||||
already-running metrics-server and should use `--metrics-run-id` matching the
|
||||
target endpoint's Skippy run id.
|
||||
|
||||
Benchmark-managed Skippy server runs must use a release `skippy-server` build.
|
||||
Run `just release-build` before `run`, `focused-runtime`, `local-single`, or
|
||||
local split binary benchmarks, and use `target/release/skippy-server` (the
|
||||
SkippyBench default). Do not use `target/debug/skippy-server` for performance or
|
||||
full-corpus validation; SkippyBench rejects that path because debug builds can
|
||||
create false timeout and throughput failures.
|
||||
|
||||
## Current Repo Shape
|
||||
|
||||
|
|
@ -28,3 +40,117 @@ cargo test -p mesh-llm-host-runtime --lib inference::skippy
|
|||
|
||||
When benchmark harnesses are imported, keep reporting separate from request-path
|
||||
serving. Stage runtimes emit telemetry; benchmark/report tooling owns reports.
|
||||
|
||||
## External Agent Evals
|
||||
|
||||
Use `skippy-bench eval` for external agent/coding benchmark harnesses. The
|
||||
local SkippyBench corpora are for runtime behavior, cache behavior, transport
|
||||
stress, and perf regression; they are not the source of agent benchmark claims.
|
||||
|
||||
Core pack:
|
||||
|
||||
```bash
|
||||
skippy-bench eval list
|
||||
skippy-bench eval info terminal-bench
|
||||
skippy-bench eval sync --pack core
|
||||
skippy-bench eval doctor
|
||||
skippy-bench eval run speed-bench \
|
||||
--base-url http://127.0.0.1:9337/v1 \
|
||||
--model org/repo:Q4_K_M \
|
||||
--endpoint-concurrency 1 \
|
||||
--metrics-http http://127.0.0.1:18080 \
|
||||
--metrics-run-id run-local-qwen
|
||||
```
|
||||
|
||||
`--timeout-secs` is passed to the native harness as its request/task timeout
|
||||
where supported. It is not a full-run dataset limit. Use
|
||||
`--harness-timeout-secs` only when you need a hard wall-clock cap for an
|
||||
operator/debug run; omit it for canonical full-dataset validation.
|
||||
`--endpoint-concurrency` must match the target endpoint's
|
||||
`serve-openai --generation-concurrency` value. SkippyBench keeps native harness
|
||||
request concurrency equal to that value; adapter-specific request concurrency
|
||||
overrides such as `SWE_BENCH_PRO_NUM_WORKERS` and
|
||||
`MCP_ATLAS_COMPLETION_CONCURRENCY` must match it or `eval run` fails before
|
||||
starting the upstream harness. Do not run multiple LLM workers against a
|
||||
single-lane Skippy endpoint when validating full corpora.
|
||||
|
||||
Core eval ids:
|
||||
|
||||
- `speed-bench` — llama.cpp SPEED-Bench client for OpenAI-compatible serving
|
||||
latency/throughput. Run the upstream qualitative benchmark across all
|
||||
categories with no Skippy-owned sample limit.
|
||||
- `terminal-bench` — Terminal-Bench CLI via `terminal-bench-core==0.1.1`.
|
||||
- `swe-bench-pro` — Scale SWE-Bench Pro OS repo; uses the upstream data and
|
||||
SWE-agent patch generation/evaluation flow rather than a Skippy-owned mini
|
||||
benchmark.
|
||||
- `mcp-atlas` — Scale MCP-Atlas native harness. `eval run` starts the
|
||||
MCP agent environment and completion service when their localhost ports are
|
||||
not already live, then runs the upstream completion script with `--no-filter`
|
||||
so all Hugging Face dataset rows are attempted, plus the upstream scoring
|
||||
path, without Skippy-specific task limits or `tool_choice` overrides.
|
||||
|
||||
Use-case routing:
|
||||
|
||||
| Need | Eval | Why |
|
||||
|---|---|---|
|
||||
| OpenAI-compatible serving latency, tok/s, and full SPEED-Bench traffic | `speed-bench` | Native SPEED-Bench client over the upstream dataset selection. |
|
||||
| Terminal agent behavior, shell/task execution, Docker sandbox readiness | `terminal-bench` | Exercises an agent loop that has to operate in a real terminal task environment. |
|
||||
| Coding-agent patch generation and issue-resolution style prompts | `swe-bench-pro` | Uses upstream SWE-agent instance generation, patch gathering, and `swe_bench_pro_eval.py`. |
|
||||
| MCP tool-use benchmark flow | `mcp-atlas` | Uses upstream MCP-Atlas completion and scoring scripts with the full Hugging Face dataset. |
|
||||
| Cache, runtime, transport, split, or mesh performance regression | Built-in SkippyBench `run`, `focused-runtime`, `local-single`, or `chat-corpus` | These are Skippy/runtime benchmarks, not external agent-quality claims. |
|
||||
|
||||
Optional future packs are intentionally not wired yet:
|
||||
|
||||
- `repo-generation`: NL2RepoBench.
|
||||
- `tool-expanded`: Toolathlon / Tool-Decathlon.
|
||||
|
||||
Keep `sync`/`install` opt-in. Do not make normal `just build` or `cargo build`
|
||||
download external harnesses, datasets, or Docker images.
|
||||
|
||||
Terminal-Bench should be installed with `uv tool install --python 3.12
|
||||
terminal-bench`; Python 3.14 currently breaks the `tb` Typer CLI. Treat Docker
|
||||
as ready only when `skippy-bench eval doctor` reports that the daemon can start
|
||||
a container; `docker info` alone is insufficient. `skippy-bench eval run`
|
||||
performs the same prerequisite checks before launching a native harness. Do not
|
||||
add Skippy-owned task filters, dataset limits, compatibility shims,
|
||||
response-format substitutions, or `tool_choice` overrides to external evals
|
||||
unless the user explicitly asks for a noncanonical experiment.
|
||||
|
||||
For MCP-Atlas scoring, the wrapper defaults `EVAL_LLM_MODEL`,
|
||||
`EVAL_LLM_BASE_URL`, and `EVAL_LLM_API_KEY` to the same local endpoint/model
|
||||
used for completion, while preserving caller-provided `EVAL_LLM_*` overrides
|
||||
for judge-model runs. When validating with a very small local Skippy model, run
|
||||
completion against the normal compatibility endpoint and point `EVAL_LLM_*` at
|
||||
a separate strict structured-output scorer endpoint, for example a second
|
||||
`skippy-server serve-openai --openai-guardrails enforce` process; do not patch
|
||||
or post-process the MCP scorer. For resumed operator runs, set
|
||||
`MCP_ATLAS_COMPLETION_OUTPUT_NAME` to an existing upstream
|
||||
`completion_results/*.csv` basename so the native completion script can reuse
|
||||
its own processed-row skip behavior, and use `MCP_ATLAS_SCORE_CONCURRENCY` for
|
||||
the scorer's native `--concurrency` setting.
|
||||
|
||||
For SWE-Bench Pro, the wrapper defaults to the official Docker image namespace
|
||||
with local Docker deployment and local Docker evaluation so the core pack can
|
||||
run without Modal credentials. It still runs upstream
|
||||
`helper_code/generate_sweagent_instances.py` for the full dataset, then supplies
|
||||
SWE-agent with a native `expert_file` instance file for local Docker platform,
|
||||
entrypoint settings and SWE-agent's standalone Python/SWE-Rex Docker runtime.
|
||||
Local Docker runs install SWE-agent into a dedicated venv and default
|
||||
`SWE_BENCH_PRO_SWEREX_SPEC` to `swe-rex[modal]==1.4.0`, which keeps the native
|
||||
SWE-ReX Docker runtime but includes the upstream
|
||||
`python:3.11.9-slim-bookworm` builder fix. Modal remains an explicit
|
||||
environment override and uses the Scale SWE-ReX patch flow. Some official
|
||||
SWE-Pro base images point pip at an unavailable localhost package mirror; local
|
||||
Docker runs default `SWE_BENCH_PRO_SWEREX_PIP_INDEX_URL` to
|
||||
`https://pypi.org/simple` for the derived-image SWE-ReX install. Use
|
||||
`SWE_BENCH_PRO_PARSE_FUNCTION=thought_action` for local OpenAI-compatible models
|
||||
that do not emit OpenAI tool calls; this is the upstream SWE-agent local-model
|
||||
path, not a Skippy dataset or harness rewrite.
|
||||
|
||||
For TTFT/FTTT, use metrics-server correlation rather than harness-only timing.
|
||||
`skippy-bench eval run` and `skippy-bench chat-corpus` create/finalize a
|
||||
metrics-server run and fail if the metrics report cannot be exported. The
|
||||
target endpoint must be emitting OTLP for the same run id. Debug telemetry is
|
||||
required for per-token spans such as `stage.openai_decode_token`; without it,
|
||||
the JSON report will still include a telemetry block explaining why TTFT/FTTT
|
||||
was unavailable.
|
||||
|
|
|
|||
2
Cargo.lock
generated
2
Cargo.lock
generated
|
|
@ -7376,6 +7376,8 @@ version = "0.72.1"
|
|||
dependencies = [
|
||||
"anyhow",
|
||||
"clap",
|
||||
"csv",
|
||||
"dirs",
|
||||
"model-artifact",
|
||||
"model-hf",
|
||||
"model-ref",
|
||||
|
|
|
|||
|
|
@ -52,7 +52,22 @@ impl StructuredOutputSpec {
|
|||
}
|
||||
|
||||
fn validate_supported_schema(schema: &Value) -> Result<(), UnsupportedStructuredSchema> {
|
||||
let definitions = schema.get("$defs").and_then(Value::as_object);
|
||||
validate_supported_schema_inner(schema, definitions, 0)
|
||||
}
|
||||
|
||||
fn validate_supported_schema_inner(
|
||||
schema: &Value,
|
||||
definitions: Option<&Map<String, Value>>,
|
||||
depth: usize,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
if depth > 32 {
|
||||
return Err(UnsupportedStructuredSchema);
|
||||
}
|
||||
let object = schema.as_object().ok_or(UnsupportedStructuredSchema)?;
|
||||
if let Some(resolved) = resolve_schema_ref(object, definitions) {
|
||||
return validate_supported_schema_inner(resolved?, definitions, depth + 1);
|
||||
}
|
||||
let schema_type = object
|
||||
.get("type")
|
||||
.and_then(Value::as_str)
|
||||
|
|
@ -61,8 +76,8 @@ fn validate_supported_schema(schema: &Value) -> Result<(), UnsupportedStructured
|
|||
reject_unsupported_keywords(object)?;
|
||||
|
||||
match schema_type {
|
||||
"object" => validate_object_schema(object),
|
||||
"array" => validate_array_schema(object),
|
||||
"object" => validate_object_schema(object, definitions, depth),
|
||||
"array" => validate_array_schema(object, definitions, depth),
|
||||
"string" | "number" | "integer" | "boolean" | "null" => validate_scalar_schema(object),
|
||||
_ => Err(UnsupportedStructuredSchema),
|
||||
}
|
||||
|
|
@ -72,7 +87,6 @@ fn reject_unsupported_keywords(
|
|||
object: &Map<String, Value>,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
const UNSUPPORTED_KEYS: &[&str] = &[
|
||||
"$ref",
|
||||
"allOf",
|
||||
"anyOf",
|
||||
"const",
|
||||
|
|
@ -94,7 +108,12 @@ fn reject_unsupported_keywords(
|
|||
}
|
||||
}
|
||||
|
||||
fn validate_object_schema(object: &Map<String, Value>) -> Result<(), UnsupportedStructuredSchema> {
|
||||
fn validate_object_schema(
|
||||
object: &Map<String, Value>,
|
||||
definitions: Option<&Map<String, Value>>,
|
||||
depth: usize,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
validate_definitions(object, definitions, depth)?;
|
||||
let properties = match object.get("properties") {
|
||||
Some(Value::Object(properties)) => Some(properties),
|
||||
Some(_) => return Err(UnsupportedStructuredSchema),
|
||||
|
|
@ -116,18 +135,23 @@ fn validate_object_schema(object: &Map<String, Value>) -> Result<(), Unsupported
|
|||
}
|
||||
if let Some(properties) = properties {
|
||||
for schema in properties.values() {
|
||||
validate_supported_schema(schema)?;
|
||||
validate_supported_schema_inner(schema, definitions, depth + 1)?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_array_schema(object: &Map<String, Value>) -> Result<(), UnsupportedStructuredSchema> {
|
||||
fn validate_array_schema(
|
||||
object: &Map<String, Value>,
|
||||
definitions: Option<&Map<String, Value>>,
|
||||
depth: usize,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
validate_definitions(object, definitions, depth)?;
|
||||
let items = object.get("items").ok_or(UnsupportedStructuredSchema)?;
|
||||
if items.is_array() {
|
||||
return Err(UnsupportedStructuredSchema);
|
||||
}
|
||||
validate_supported_schema(items)
|
||||
validate_supported_schema_inner(items, definitions, depth + 1)
|
||||
}
|
||||
|
||||
fn validate_scalar_schema(object: &Map<String, Value>) -> Result<(), UnsupportedStructuredSchema> {
|
||||
|
|
@ -143,14 +167,30 @@ fn validate_payload_against_schema(
|
|||
schema: &Value,
|
||||
payload: &Value,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
let definitions = schema.get("$defs").and_then(Value::as_object);
|
||||
validate_payload_against_schema_inner(schema, payload, definitions, 0)
|
||||
}
|
||||
|
||||
fn validate_payload_against_schema_inner(
|
||||
schema: &Value,
|
||||
payload: &Value,
|
||||
definitions: Option<&Map<String, Value>>,
|
||||
depth: usize,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
if depth > 32 {
|
||||
return Err(UnsupportedStructuredSchema);
|
||||
}
|
||||
let object = schema.as_object().ok_or(UnsupportedStructuredSchema)?;
|
||||
if let Some(resolved) = resolve_schema_ref(object, definitions) {
|
||||
return validate_payload_against_schema_inner(resolved?, payload, definitions, depth + 1);
|
||||
}
|
||||
match object
|
||||
.get("type")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or(UnsupportedStructuredSchema)?
|
||||
{
|
||||
"object" => validate_object_payload(object, payload),
|
||||
"array" => validate_array_payload(object, payload),
|
||||
"object" => validate_object_payload(object, payload, definitions, depth),
|
||||
"array" => validate_array_payload(object, payload, definitions, depth),
|
||||
"string" => payload
|
||||
.as_str()
|
||||
.map(|_| ())
|
||||
|
|
@ -179,6 +219,8 @@ fn validate_payload_against_schema(
|
|||
fn validate_object_payload(
|
||||
schema: &Map<String, Value>,
|
||||
payload: &Value,
|
||||
definitions: Option<&Map<String, Value>>,
|
||||
depth: usize,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
let payload = payload.as_object().ok_or(UnsupportedStructuredSchema)?;
|
||||
let properties = schema
|
||||
|
|
@ -203,7 +245,7 @@ fn validate_object_payload(
|
|||
.unwrap_or(true);
|
||||
for (key, value) in payload {
|
||||
if let Some(property_schema) = properties.get(key) {
|
||||
validate_payload_against_schema(property_schema, value)?;
|
||||
validate_payload_against_schema_inner(property_schema, value, definitions, depth + 1)?;
|
||||
} else if !allow_additional {
|
||||
return Err(UnsupportedStructuredSchema);
|
||||
}
|
||||
|
|
@ -214,11 +256,135 @@ fn validate_object_payload(
|
|||
fn validate_array_payload(
|
||||
schema: &Map<String, Value>,
|
||||
payload: &Value,
|
||||
definitions: Option<&Map<String, Value>>,
|
||||
depth: usize,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
let payload = payload.as_array().ok_or(UnsupportedStructuredSchema)?;
|
||||
let item_schema = schema.get("items").ok_or(UnsupportedStructuredSchema)?;
|
||||
for item in payload {
|
||||
validate_payload_against_schema(item_schema, item)?;
|
||||
validate_payload_against_schema_inner(item_schema, item, definitions, depth + 1)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_definitions(
|
||||
object: &Map<String, Value>,
|
||||
definitions: Option<&Map<String, Value>>,
|
||||
depth: usize,
|
||||
) -> Result<(), UnsupportedStructuredSchema> {
|
||||
let Some(Value::Object(local_definitions)) = object.get("$defs") else {
|
||||
return Ok(());
|
||||
};
|
||||
for definition in local_definitions.values() {
|
||||
validate_supported_schema_inner(definition, definitions, depth + 1)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn resolve_schema_ref<'a>(
|
||||
object: &'a Map<String, Value>,
|
||||
definitions: Option<&'a Map<String, Value>>,
|
||||
) -> Option<Result<&'a Value, UnsupportedStructuredSchema>> {
|
||||
let reference = object.get("$ref")?;
|
||||
let Some(reference) = reference.as_str() else {
|
||||
return Some(Err(UnsupportedStructuredSchema));
|
||||
};
|
||||
if object
|
||||
.keys()
|
||||
.any(|key| key != "$ref" && key != "description" && key != "title")
|
||||
{
|
||||
return Some(Err(UnsupportedStructuredSchema));
|
||||
}
|
||||
let Some(name) = reference.strip_prefix("#/$defs/") else {
|
||||
return Some(Err(UnsupportedStructuredSchema));
|
||||
};
|
||||
let name = decode_json_pointer_segment(name);
|
||||
Some(name.and_then(|name| {
|
||||
definitions
|
||||
.and_then(|definitions| definitions.get(&name))
|
||||
.ok_or(UnsupportedStructuredSchema)
|
||||
}))
|
||||
}
|
||||
|
||||
fn decode_json_pointer_segment(segment: &str) -> Result<String, UnsupportedStructuredSchema> {
|
||||
let mut decoded = String::with_capacity(segment.len());
|
||||
let mut chars = segment.chars();
|
||||
while let Some(ch) = chars.next() {
|
||||
if ch != '~' {
|
||||
decoded.push(ch);
|
||||
continue;
|
||||
}
|
||||
match chars.next() {
|
||||
Some('0') => decoded.push('~'),
|
||||
Some('1') => decoded.push('/'),
|
||||
_ => return Err(UnsupportedStructuredSchema),
|
||||
}
|
||||
}
|
||||
Ok(decoded)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use serde_json::json;
|
||||
|
||||
use super::StructuredOutputSpec;
|
||||
|
||||
#[test]
|
||||
fn json_schema_supports_local_defs_refs() {
|
||||
let response_format = json!({
|
||||
"type": "json_schema",
|
||||
"json_schema": {
|
||||
"name": "CommandBatchResponse",
|
||||
"strict": true,
|
||||
"schema": {
|
||||
"$defs": {
|
||||
"Command": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"keystrokes": {"type": "string"},
|
||||
"is_blocking": {"type": "boolean"},
|
||||
"timeout_sec": {"type": "number"}
|
||||
},
|
||||
"required": ["keystrokes", "is_blocking", "timeout_sec"],
|
||||
"additionalProperties": false
|
||||
}
|
||||
},
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"state_analysis": {"type": "string"},
|
||||
"commands": {
|
||||
"type": "array",
|
||||
"items": {"$ref": "#/$defs/Command"}
|
||||
},
|
||||
"is_task_complete": {"type": "boolean"}
|
||||
},
|
||||
"required": ["state_analysis", "commands", "is_task_complete"],
|
||||
"additionalProperties": false
|
||||
}
|
||||
}
|
||||
});
|
||||
let spec =
|
||||
StructuredOutputSpec::from_response_format_object(response_format.as_object().unwrap())
|
||||
.expect("local refs should be supported");
|
||||
|
||||
spec.validate_payload(&json!({
|
||||
"state_analysis": "ready",
|
||||
"commands": [{
|
||||
"keystrokes": "ls\n",
|
||||
"is_blocking": true,
|
||||
"timeout_sec": 1
|
||||
}],
|
||||
"is_task_complete": false
|
||||
}))
|
||||
.expect("payload should validate through ref");
|
||||
|
||||
assert!(
|
||||
spec.validate_payload(&json!({
|
||||
"state_analysis": "ready",
|
||||
"commands": [{"keystrokes": "ls\n"}],
|
||||
"is_task_complete": false
|
||||
}))
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,7 +28,10 @@ pub use mesh_llm_guardrails::{
|
|||
CompactionConfig, CompactionDecision, CompactionOverride, CompactionReport, MESH_COMPACT_FIELD,
|
||||
MESH_RESPOND_TOOL_NAME,
|
||||
};
|
||||
pub use policy::{GuardrailMode, GuardrailPolicy, GuardrailPolicyHandle, StreamingGuardrailMode};
|
||||
pub use policy::{
|
||||
GuardrailMode, GuardrailPolicy, GuardrailPolicyHandle, RetryExhaustionMode,
|
||||
StreamingGuardrailMode,
|
||||
};
|
||||
pub use telemetry::GuardrailTelemetrySink;
|
||||
|
||||
use self::{
|
||||
|
|
|
|||
|
|
@ -452,6 +452,12 @@ fn classify_tool_call_value(
|
|||
parser_stage: GuardrailParserStage,
|
||||
finish_reason: Option<FinishReason>,
|
||||
) -> ClassifiedGuardrailResponse {
|
||||
if let Some(classified) =
|
||||
classify_direct_structured_payload(prepared, value, parser_stage, finish_reason)
|
||||
{
|
||||
return classified;
|
||||
}
|
||||
|
||||
let allowed_real_tools = allowed_real_tool_names(prepared);
|
||||
let allowed_backend_tools = allowed_backend_tool_names(prepared);
|
||||
let raw_tool_calls = match raw_tool_calls_from_value(value) {
|
||||
|
|
@ -672,6 +678,37 @@ fn classify_tool_call_value(
|
|||
}
|
||||
}
|
||||
|
||||
fn classify_direct_structured_payload(
|
||||
prepared: &PreparedGuardrailRequest,
|
||||
value: &Value,
|
||||
parser_stage: GuardrailParserStage,
|
||||
finish_reason: Option<FinishReason>,
|
||||
) -> Option<ClassifiedGuardrailResponse> {
|
||||
if prepared.state.request_contract.has_real_tools() {
|
||||
return None;
|
||||
}
|
||||
let spec = prepared.state.request_contract.structured_output_spec()?;
|
||||
value.as_object()?;
|
||||
let valid_payload = spec.validate_payload(value).is_ok();
|
||||
Some(ClassifiedGuardrailResponse {
|
||||
category: if valid_payload {
|
||||
GuardrailResponseCategory::ValidSyntheticStructured
|
||||
} else {
|
||||
GuardrailResponseCategory::InvalidStructuredPayload
|
||||
},
|
||||
parser_stage,
|
||||
visible_content: None,
|
||||
tool_calls: None,
|
||||
synthetic_text: None,
|
||||
structured_payload: if valid_payload {
|
||||
Some(value.clone())
|
||||
} else {
|
||||
None
|
||||
},
|
||||
finish_reason,
|
||||
})
|
||||
}
|
||||
|
||||
fn raw_tool_calls_from_value(value: &Value) -> Option<Vec<&Value>> {
|
||||
match value {
|
||||
Value::Array(entries) => Some(entries.iter().collect()),
|
||||
|
|
|
|||
|
|
@ -386,6 +386,52 @@ fn synthetic_structured_classifies_when_allowed() {
|
|||
assert_eq!(classified.structured_payload, Some(json!({"answer": 42})));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn direct_structured_json_object_classifies_when_allowed() {
|
||||
let engine = GuardrailEngine::new(enforce_policy());
|
||||
let prepared = prepared_text_request(
|
||||
&engine,
|
||||
json!({
|
||||
"model": "Qwen3-8B-Q4_K_M",
|
||||
"messages": [{"role": "user", "content": "json"}],
|
||||
"response_format": supported_json_schema_response_format()
|
||||
}),
|
||||
);
|
||||
let response = response_with_content("Qwen3-8B-Q4_K_M", r#"{"answer":42}"#);
|
||||
|
||||
let classified = engine.classify_response(&prepared, &response);
|
||||
|
||||
assert_eq!(
|
||||
classified.category,
|
||||
GuardrailResponseCategory::ValidSyntheticStructured
|
||||
);
|
||||
assert_eq!(classified.parser_stage, GuardrailParserStage::JsonExact);
|
||||
assert_eq!(classified.structured_payload, Some(json!({"answer": 42})));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fenced_direct_structured_json_object_classifies_when_allowed() {
|
||||
let engine = GuardrailEngine::new(enforce_policy());
|
||||
let prepared = prepared_text_request(
|
||||
&engine,
|
||||
json!({
|
||||
"model": "Qwen3-8B-Q4_K_M",
|
||||
"messages": [{"role": "user", "content": "json"}],
|
||||
"response_format": supported_json_schema_response_format()
|
||||
}),
|
||||
);
|
||||
let response = response_with_content("Qwen3-8B-Q4_K_M", "```json\n{\"answer\":42}\n```");
|
||||
|
||||
let classified = engine.classify_response(&prepared, &response);
|
||||
|
||||
assert_eq!(
|
||||
classified.category,
|
||||
GuardrailResponseCategory::ValidSyntheticStructured
|
||||
);
|
||||
assert_eq!(classified.parser_stage, GuardrailParserStage::JsonFenced);
|
||||
assert_eq!(classified.structured_payload, Some(json!({"answer": 42})));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_structured_payload_classifies_without_leaking_arguments() {
|
||||
let engine = GuardrailEngine::new(enforce_policy());
|
||||
|
|
@ -1019,6 +1065,40 @@ async fn valid_structured_payload_becomes_json_assistant_text() {
|
|||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn direct_structured_payload_becomes_json_assistant_text() {
|
||||
let backend = Arc::new(SequencedBackend::new(vec![Ok(response_with_content(
|
||||
"Qwen3-8B-Q4_K_M",
|
||||
"```json\n{\"answer\":42}\n```",
|
||||
))]));
|
||||
let guarded = GuardedOpenAiBackend::new(
|
||||
backend,
|
||||
GuardrailPolicy {
|
||||
mode: GuardrailMode::Enforce,
|
||||
apply_to_all_models: true,
|
||||
..GuardrailPolicy::default()
|
||||
},
|
||||
);
|
||||
let request: ChatCompletionRequest = serde_json::from_value(json!({
|
||||
"model": "Qwen3-8B-Q4_K_M",
|
||||
"messages": [{"role": "user", "content": "json"}],
|
||||
"response_format": supported_json_schema_response_format()
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
let response = guarded.chat_completion(request).await.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
response.choices[0].message.content.as_deref(),
|
||||
Some("{\"answer\":42}")
|
||||
);
|
||||
assert!(response.choices[0].message.tool_calls.is_none());
|
||||
assert_eq!(
|
||||
response.choices[0].finish_reason,
|
||||
Some(crate::common::FinishReason::Stop)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn invalid_structured_payload_retries_then_exhaustion_error() {
|
||||
let backend = Arc::new(SequencedBackend::new(vec![
|
||||
|
|
|
|||
|
|
@ -32,7 +32,8 @@ pub use errors::{OpenAiError, OpenAiErrorKind, already_openai_error, map_upstrea
|
|||
pub use guardrails::{
|
||||
CompactingOpenAiBackend, CompactionConfig, CompactionDecision, CompactionOverride,
|
||||
CompactionReport, GuardedOpenAiBackend, GuardrailMode, GuardrailPolicy, GuardrailPolicyHandle,
|
||||
GuardrailTelemetrySink, MESH_COMPACT_FIELD, MESH_RESPOND_TOOL_NAME, StreamingGuardrailMode,
|
||||
GuardrailTelemetrySink, MESH_COMPACT_FIELD, MESH_RESPOND_TOOL_NAME, RetryExhaustionMode,
|
||||
StreamingGuardrailMode,
|
||||
};
|
||||
pub use hooks::{
|
||||
ChatHookAction, ChatHookOutcome, ChatMediaKind, ChatMediaRef, GenerationHookSignals,
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ version.workspace = true
|
|||
[dependencies]
|
||||
anyhow.workspace = true
|
||||
clap.workspace = true
|
||||
csv = "1.4.0"
|
||||
dirs = "6.0.0"
|
||||
skippy-protocol = { path = "../skippy-protocol" }
|
||||
skippy-runtime = { path = "../skippy-runtime" }
|
||||
skippy-topology = { path = "../skippy-topology" }
|
||||
|
|
|
|||
|
|
@ -52,16 +52,41 @@ skippy-bench local-single --model-path model.gguf --model-id org/repo:Q4_K_M
|
|||
skippy-bench local-split-binary --model-path model.gguf --model-id org/repo:Q4_K_M
|
||||
skippy-bench local-split-compare --model-path model.gguf --model-id org/repo:Q4_K_M
|
||||
skippy-bench local-split-chain-binary --model-path model.gguf --model-id org/repo:Q4_K_M
|
||||
skippy-bench chat-corpus --base-url http://127.0.0.1:9337/v1 --model org/repo:Q4_K_M --prompt-corpus target/bench-corpora/smoke/corpus.jsonl --max-tokens 64 --stream
|
||||
skippy-bench chat-corpus --base-url http://127.0.0.1:9337/v1 --model org/repo:Q4_K_M --metrics-http http://127.0.0.1:18080 --metrics-run-id run-local-qwen --prompt-corpus target/bench-corpora/smoke/corpus.jsonl --max-tokens 64 --stream
|
||||
skippy-bench token-lengths --model-path model.gguf --prompt-corpus target/bench-corpora/long/corpus.jsonl --ctx-size 8192 --generation-limit 512 --output-tsv target/bench-corpora/long/prompt-lengths.tsv
|
||||
skippy-bench focused-runtime --schema-smoke --hosts host-a,host-b --splits 1 --layer-end 2
|
||||
skippy-bench eval list
|
||||
skippy-bench eval sync --pack core
|
||||
skippy-bench eval run speed-bench --base-url http://127.0.0.1:9337/v1 --model org/repo:Q4_K_M --metrics-http http://127.0.0.1:18080 --metrics-run-id run-local-qwen
|
||||
```
|
||||
|
||||
Benchmark-managed Skippy server runs require a release `skippy-server` binary.
|
||||
Run `just release-build` before `run`, `focused-runtime`, `local-single`, or
|
||||
local split binary benchmarks. These commands default to
|
||||
`target/release/skippy-server` and reject `target/debug/skippy-server` because
|
||||
debug builds distort throughput and timeout behavior.
|
||||
|
||||
The old standalone `kv-stage-integration` and `kv-hit-regression` commands are
|
||||
intentionally absent. Mesh does not carry the legacy standalone cache sidecar
|
||||
path; exact cache work should be reintroduced through the embedded runtime and
|
||||
mesh-owned lifecycle.
|
||||
|
||||
Every reportable benchmark path must use metrics-server. `run`, `focused-runtime`,
|
||||
and `local-single` launch their own collector by default through
|
||||
`--metrics-server-bin`, `--metrics-http-addr`, and `--metrics-otlp-grpc-addr`.
|
||||
Endpoint-driving benchmarks (`chat-corpus` and `eval run`) require an existing
|
||||
metrics-server at `--metrics-http` and fail before running traffic if the run
|
||||
cannot be created. For correlated server-side TTFT/FTTT, launch the target
|
||||
Skippy/OpenAI endpoint so it exports OTLP to that collector with the same
|
||||
`--metrics-run-id`.
|
||||
|
||||
```bash
|
||||
target/debug/metrics-server serve \
|
||||
--db /tmp/skippy-bench-metrics.duckdb \
|
||||
--http-addr 127.0.0.1:18080 \
|
||||
--otlp-grpc-addr 127.0.0.1:14317
|
||||
```
|
||||
|
||||
Benchmark reports carry `model_identity` beside the public `model_id`. The
|
||||
public id is a coordinate such as `org/repo:Q4_K_M`; when the model path comes
|
||||
from the Hugging Face cache, that resolved identity is used for stage configs
|
||||
|
|
@ -75,6 +100,131 @@ can compare baseline K/V cache storage against runtime-supported package candida
|
|||
such as `q8_0`. The experimental TCQ/TurboQuant lane is intentionally not
|
||||
compiled into mesh-llm.
|
||||
|
||||
## External Agent Evals
|
||||
|
||||
`skippy-bench eval` manages external benchmark harnesses and points them at an
|
||||
already-running OpenAI-compatible Skippy or mesh endpoint. External evals are
|
||||
for agent/coding benchmark claims; the local corpora below remain runtime,
|
||||
cache, routing, and transport stress traffic.
|
||||
|
||||
The current core pack is:
|
||||
|
||||
| Eval id | External harness | Default run |
|
||||
|---|---|---|
|
||||
| `speed-bench` | llama.cpp `tools/server/bench/speed-bench` | Native SPEED-Bench qualitative run across all categories, no sample limit, `--osl 1024` |
|
||||
| `terminal-bench` | Terminal-Bench CLI (`tb`) | `terminal-bench-core==0.1.1`, Terminus agent, no task-id filter |
|
||||
| `swe-bench-pro` | Scale SWE-Bench Pro OS repo | Upstream SWE-agent patch generation, patch gathering, and `swe_bench_pro_eval.py` |
|
||||
| `mcp-atlas` | Scale MCP-Atlas repo | Native MCP-Atlas completion script with upstream `--no-filter`, plus scoring through auto-started MCP services |
|
||||
|
||||
```bash
|
||||
skippy-bench eval list
|
||||
skippy-bench eval info terminal-bench
|
||||
skippy-bench eval sync --pack core
|
||||
skippy-bench eval doctor
|
||||
skippy-bench eval run terminal-bench \
|
||||
--base-url http://127.0.0.1:9337/v1 \
|
||||
--model org/repo:Q4_K_M \
|
||||
--metrics-http http://127.0.0.1:18080 \
|
||||
--metrics-run-id run-local-qwen
|
||||
```
|
||||
|
||||
`--timeout-secs` is forwarded to native harnesses as their request/task timeout
|
||||
where supported. It is not a SkippyBench dataset limit and does not cap full
|
||||
canonical runs. Use `--harness-timeout-secs` only when an operator wants a hard
|
||||
wall-clock cap around the native harness process for debugging or CI guardrails.
|
||||
`--endpoint-concurrency` declares the target OpenAI endpoint's generation
|
||||
concurrency and defaults to `1`. SkippyBench keeps each external harness's LLM
|
||||
request concurrency equal to that value. If an adapter-specific request
|
||||
concurrency override such as `SWE_BENCH_PRO_NUM_WORKERS` or
|
||||
`MCP_ATLAS_COMPLETION_CONCURRENCY` is set to a different value, `eval run`
|
||||
fails before launching the native harness.
|
||||
|
||||
`sync` clones or installs the external harnesses into
|
||||
`~/.cache/mesh-llm/skippy-bench/harnesses/` by default. Use `--cache-root` to
|
||||
override that location. Use `--dry-run` with `sync` or `run` to inspect the
|
||||
commands without cloning, pulling Docker images, or launching a benchmark.
|
||||
Before launching native harness traffic, `eval run` enforces the same required
|
||||
tool checks as `eval doctor`, including Docker container-start readiness for
|
||||
Docker-backed evals.
|
||||
Terminal-Bench is installed through `uv tool install --python 3.12` because the
|
||||
current `tb` CLI is not compatible with Python 3.14. `eval doctor` checks that
|
||||
Docker's daemon is reachable and can start a tiny container, not just that the
|
||||
`docker` CLI exists or that `docker info` returns.
|
||||
MCP-Atlas starts its Docker agent environment and Python completion service
|
||||
when ports `1984` and `3000` are not already reachable, waits for readiness,
|
||||
and cleans up only the services that the run started. The adapter runs the
|
||||
upstream completion script with `--no-filter` so all Hugging Face dataset rows
|
||||
are attempted, and without Skippy-specific task limits or `tool_choice`
|
||||
overrides. By default, the MCP-Atlas scorer uses the same local
|
||||
OpenAI-compatible endpoint/model as the completion run; set `EVAL_LLM_MODEL`,
|
||||
`EVAL_LLM_BASE_URL`, and `EVAL_LLM_API_KEY` to use a separate judge model. For
|
||||
small local Skippy validation models, keep the completion endpoint in normal
|
||||
compatibility mode and point the scorer override at a strict structured-output
|
||||
endpoint, for example a second `skippy-server serve-openai
|
||||
--openai-guardrails enforce` process. The adapter still uses the native scorer
|
||||
and does not rewrite score data. For operator resumes, set
|
||||
`MCP_ATLAS_COMPLETION_OUTPUT_NAME` to an existing upstream
|
||||
`completion_results/*.csv` basename so the native completion script can reuse
|
||||
its own resume behavior, and set `MCP_ATLAS_SCORE_CONCURRENCY` to the upstream
|
||||
scorer's `--concurrency` value.
|
||||
SWE-Bench Pro defaults to the official Docker image namespace (`jefzda`) with
|
||||
local Docker deployment and local Docker evaluation so the core pack can run
|
||||
without Modal credentials. The adapter first runs upstream
|
||||
`helper_code/generate_sweagent_instances.py` for the full dataset, then feeds
|
||||
SWE-agent a native `expert_file` instance file so local Docker can set the
|
||||
official image platform, clear image entrypoints, and use SWE-agent's
|
||||
standalone Python/SWE-Rex Docker runtime. Local Docker runs install SWE-agent
|
||||
into a dedicated venv and default `SWE_BENCH_PRO_SWEREX_SPEC` to
|
||||
`swe-rex[modal]==1.4.0`, which preserves SWE-ReX's native Docker runtime while
|
||||
using the upstream `python:3.11.9-slim-bookworm` builder fix. Modal runs keep
|
||||
the Scale SWE-Rex patch flow. Some SWE-Pro base images carry a pip index config
|
||||
for an unavailable localhost mirror, so the local Docker adapter defaults
|
||||
`SWE_BENCH_PRO_SWEREX_PIP_INDEX_URL` to `https://pypi.org/simple` for the
|
||||
derived-image SWE-ReX install step. Override
|
||||
`SWE_BENCH_PRO_DOCKERHUB_USERNAME`,
|
||||
`SWE_BENCH_PRO_DOCKER_PLATFORM`, `SWE_BENCH_PRO_DEPLOYMENT_TYPE`,
|
||||
`SWE_BENCH_PRO_NUM_WORKERS`, `SWE_BENCH_PRO_EVAL_WORKERS`, or
|
||||
`SWE_BENCH_PRO_PARSE_FUNCTION`. Use `SWE_BENCH_PRO_PYTHON` to choose the
|
||||
SWE-agent venv interpreter and `SWE_BENCH_PRO_SWEREX_SPEC` to override the
|
||||
SWE-ReX package spec. Use `SWE_BENCH_PRO_SWEREX_PIP_INDEX_URL` to choose the
|
||||
package index used inside SWE-ReX derived Docker images. Set
|
||||
`SWE_BENCH_PRO_PARSE_FUNCTION=thought_action` for local OpenAI-compatible
|
||||
models that do not emit OpenAI tool calls; this is the upstream SWE-agent
|
||||
local-model path. Set `SWE_BENCH_PRO_USE_LOCAL_DOCKER=0` when running the full
|
||||
harness in a different environment such as Modal.
|
||||
|
||||
Every `eval run` writes `run.json` under the run directory with command status,
|
||||
raw artifact paths, wall-clock duration, and normalized metrics where the
|
||||
harness exposes them. `speed-bench` records request counts, latency,
|
||||
prompt/completion/total token counts, prompt and completion tok/s, and draft
|
||||
acceptance rate when the server returns llama.cpp-compatible `timings`.
|
||||
SWE-Bench Pro records OpenAI usage tokens and client-side tok/s when the
|
||||
upstream flow produces them.
|
||||
Terminal-Bench records pass rate, resolved/unresolved task counts, token totals
|
||||
when the agent reports them, and raw harness artifacts. The MCP-Atlas adapter
|
||||
records wall time, raw completion CSV artifacts, the native scoring output
|
||||
directory, and CSV task row count.
|
||||
|
||||
`eval run` requires metrics-server for every external benchmark. `--metrics-http` defaults to
|
||||
`http://127.0.0.1:18080`; the command creates a metrics-server run before the
|
||||
harness starts and fails if that run cannot be created. Pass
|
||||
`--metrics-run-id` to correlate the eval with the target Skippy/OpenAI endpoint
|
||||
run id. SkippyBench finalizes and fetches
|
||||
`/v1/runs/<run-id>/report.json`, stores it as `raw/metrics-report.json`, and
|
||||
adds a `telemetry` block to `run.json`. When the target emits debug telemetry,
|
||||
SkippyBench derives TTFT/FTTT from the first request span to the first
|
||||
`stage.openai_decode_token` span, plus request and generation latency
|
||||
aggregates. If the target endpoint is not emitting the requested run id, or if
|
||||
debug token spans are disabled, the telemetry block records that status rather
|
||||
than filling misleading values.
|
||||
|
||||
Optional packs intentionally not wired yet:
|
||||
|
||||
| Future pack | Candidate |
|
||||
|---|---|
|
||||
| `repo-generation` | NL2RepoBench |
|
||||
| `tool-expanded` | Toolathlon / Tool-Decathlon |
|
||||
|
||||
## Benchmark Corpora
|
||||
|
||||
Benchmark corpora are generated from Hugging Face datasets instead of checked
|
||||
|
|
@ -137,6 +287,10 @@ The `long-context` tier keeps a much larger prompt character budget and expands
|
|||
sampled HF text into long stress packets. It is for 32k context capacity and
|
||||
transport stress only; do not substitute it for the 8k customer-readiness
|
||||
baseline or quality/speculation decisions.
|
||||
The built-in manifest intentionally excludes generic chat, math, summarization,
|
||||
SQL, and standalone function-calling sources such as OASST, Dolly, GSM8K, XSum,
|
||||
Spider, and xLAM. Agent/coding claims should use the external eval harnesses
|
||||
above rather than local prompt sampling.
|
||||
The manifest records source datasets, resolved revisions, downloaded parquet
|
||||
files, quotas, generated row counts, seed, generator path, and generator git
|
||||
commit.
|
||||
|
|
@ -179,6 +333,8 @@ customer-facing benchmark numbers after the stage topology is already running:
|
|||
skippy-bench chat-corpus \
|
||||
--base-url http://127.0.0.1:9337/v1 \
|
||||
--model org/repo:Q4_K_M \
|
||||
--metrics-http http://127.0.0.1:18080 \
|
||||
--metrics-run-id run-local-qwen \
|
||||
--prompt-corpus target/bench-corpora/long/corpus.jsonl \
|
||||
--max-tokens 512 \
|
||||
--concurrency-depth 1 \
|
||||
|
|
@ -195,6 +351,11 @@ benchmarks can exercise per-session KV or n-gram history. It records
|
|||
per-request elapsed time, streaming TTFT when `--stream` is enabled, usage
|
||||
tokens when the frontend returns them, API error codes, and aggregate
|
||||
latency/token-rate summaries.
|
||||
The command creates/finalizes a metrics-server run, writes the raw
|
||||
metrics-server report beside `--output` by default, adds a telemetry summary to
|
||||
the chat-corpus JSON report, and fails if the metrics-server report cannot be
|
||||
created. It also sends stable `x-request-id` headers so matching server spans
|
||||
can be grouped cleanly when the target endpoint exports the same run id.
|
||||
Use `--concurrency-depth` for depth sweeps; the effective frontend generation
|
||||
limit, such as `serve-openai --generation-concurrency`, must still be recorded
|
||||
beside the result.
|
||||
|
|
|
|||
|
|
@ -118,90 +118,6 @@
|
|||
"smoke": 2,
|
||||
"long": 50
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "xlam-function-calling",
|
||||
"dataset": "product-science/xlam-function-calling-60k-raw",
|
||||
"config": "default",
|
||||
"split": "train",
|
||||
"revision": "dfbd3c669354c27f2727870d39a4d86c32381448",
|
||||
"family": "structured_tool_call",
|
||||
"adapter": "xlam_tool_call",
|
||||
"routing_hint": "ngram",
|
||||
"quota": {
|
||||
"smoke": 2,
|
||||
"long": 50
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "spider-text2sql",
|
||||
"dataset": "SuperMax991/spider-text2sql",
|
||||
"config": "default",
|
||||
"split": "test",
|
||||
"revision": "890c458c18c000c0faf995aa62ee04cc91476c98",
|
||||
"family": "structured_sql",
|
||||
"adapter": "spider_sql",
|
||||
"routing_hint": "ngram",
|
||||
"quota": {
|
||||
"smoke": 2,
|
||||
"long": 50
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "oasst2-chat",
|
||||
"dataset": "OpenAssistant/oasst2",
|
||||
"config": "default",
|
||||
"split": "train",
|
||||
"revision": "179dd21fc55192153d94adb0e0ce8f69e222bf75",
|
||||
"family": "chat_open",
|
||||
"adapter": "oasst_prompt",
|
||||
"routing_hint": "baseline",
|
||||
"quota": {
|
||||
"smoke": 2,
|
||||
"long": 50
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "dolly-instruct",
|
||||
"dataset": "databricks/databricks-dolly-15k",
|
||||
"config": "default",
|
||||
"split": "train",
|
||||
"revision": "bdd27f4d94b9c1f951818a7da7fd7aeea5dbff1a",
|
||||
"family": "instruction_general",
|
||||
"adapter": "dolly_instruction",
|
||||
"routing_hint": "draft",
|
||||
"quota": {
|
||||
"smoke": 2,
|
||||
"long": 50
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "gsm8k",
|
||||
"dataset": "openai/gsm8k",
|
||||
"config": "main",
|
||||
"split": "test",
|
||||
"revision": "740312add88f781978c0658806c59bc2815b9866",
|
||||
"family": "reasoning_math",
|
||||
"adapter": "gsm8k_reasoning",
|
||||
"routing_hint": "baseline",
|
||||
"quota": {
|
||||
"smoke": 2,
|
||||
"long": 50
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "xsum",
|
||||
"dataset": "EdinburghNLP/xsum",
|
||||
"config": "default",
|
||||
"split": "test",
|
||||
"revision": "7d4d486c2f8ef850b1a11aead99b894ff3dd7da9",
|
||||
"family": "summarization",
|
||||
"adapter": "xsum_summarize",
|
||||
"routing_hint": "draft",
|
||||
"quota": {
|
||||
"smoke": 2,
|
||||
"long": 50
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,12 +1,13 @@
|
|||
use std::{
|
||||
fs,
|
||||
io::{BufRead, BufReader},
|
||||
path::{Path, PathBuf},
|
||||
sync::{
|
||||
Arc, Mutex,
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
},
|
||||
thread,
|
||||
time::{Duration, Instant},
|
||||
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
|
|
@ -14,7 +15,10 @@ use reqwest::blocking::Client;
|
|||
use serde::Serialize;
|
||||
use serde_json::{Map, Value, json};
|
||||
|
||||
use crate::cli::ChatCorpusArgs;
|
||||
use crate::{
|
||||
cli::ChatCorpusArgs,
|
||||
telemetry_report::{self, BenchTelemetry},
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct PromptCase {
|
||||
|
|
@ -29,6 +33,9 @@ struct PromptCase {
|
|||
|
||||
#[derive(Serialize)]
|
||||
struct ChatCorpusReport {
|
||||
run_id: String,
|
||||
metrics_http: String,
|
||||
metrics_run_id: String,
|
||||
base_url: String,
|
||||
model: String,
|
||||
endpoint: &'static str,
|
||||
|
|
@ -43,6 +50,7 @@ struct ChatCorpusReport {
|
|||
sampling: SamplingReport,
|
||||
results: Vec<ChatCorpusResult>,
|
||||
summary: ChatCorpusSummary,
|
||||
telemetry: BenchTelemetry,
|
||||
}
|
||||
|
||||
#[derive(Default, Serialize)]
|
||||
|
|
@ -102,6 +110,29 @@ pub fn chat_corpus(args: ChatCorpusArgs) -> Result<()> {
|
|||
}
|
||||
|
||||
let prompts = Arc::new(prompt_cases(&args)?);
|
||||
let run_id = args
|
||||
.run_id
|
||||
.clone()
|
||||
.unwrap_or_else(generate_chat_corpus_run_id);
|
||||
let metrics_run_id = args
|
||||
.metrics_run_id
|
||||
.clone()
|
||||
.unwrap_or_else(|| run_id.clone());
|
||||
let metrics_http = args.metrics_http.trim_end_matches('/').to_string();
|
||||
let metrics_report_output = metrics_report_output_path(&args, &metrics_run_id);
|
||||
let run_config = json!({
|
||||
"mode": "skippy-bench-chat-corpus",
|
||||
"run_id": &run_id,
|
||||
"model": &args.model,
|
||||
"base_url": &args.base_url,
|
||||
"stream": args.stream,
|
||||
"request_count": prompts.len(),
|
||||
"prompt_corpus": args.prompt_corpus.as_ref().map(|path| path.display().to_string()),
|
||||
"prompt_limit": args.prompt_limit,
|
||||
"max_tokens": args.max_tokens,
|
||||
"concurrency_depth": args.concurrency_depth,
|
||||
});
|
||||
telemetry_report::create_run(&metrics_http, &metrics_run_id, &run_config)?;
|
||||
let client = Client::builder()
|
||||
.timeout(Duration::from_secs(args.request_timeout_secs))
|
||||
.build()
|
||||
|
|
@ -139,8 +170,21 @@ pub fn chat_corpus(args: ChatCorpusArgs) -> Result<()> {
|
|||
.into_inner()
|
||||
.expect("results mutex poisoned");
|
||||
results.sort_by_key(|result| result.sequence);
|
||||
let telemetry_result = telemetry_report::finalize_and_collect(
|
||||
&metrics_http,
|
||||
&metrics_run_id,
|
||||
&metrics_report_output,
|
||||
);
|
||||
let telemetry_error = telemetry_result.as_ref().err().map(ToString::to_string);
|
||||
let telemetry = match telemetry_result {
|
||||
Ok(telemetry) => telemetry,
|
||||
Err(error) => telemetry_report::unavailable(&metrics_http, &metrics_run_id, &error),
|
||||
};
|
||||
|
||||
let report = ChatCorpusReport {
|
||||
run_id,
|
||||
metrics_http,
|
||||
metrics_run_id,
|
||||
base_url: args.base_url.trim_end_matches('/').to_string(),
|
||||
model: args.model.clone(),
|
||||
endpoint: "/v1/chat/completions",
|
||||
|
|
@ -165,6 +209,7 @@ pub fn chat_corpus(args: ChatCorpusArgs) -> Result<()> {
|
|||
},
|
||||
summary: summarize(&results, total_wall_ms),
|
||||
results,
|
||||
telemetry,
|
||||
};
|
||||
|
||||
let json = serde_json::to_vec_pretty(&report)?;
|
||||
|
|
@ -173,6 +218,9 @@ pub fn chat_corpus(args: ChatCorpusArgs) -> Result<()> {
|
|||
.with_context(|| format!("failed to write {}", output.display()))?;
|
||||
}
|
||||
println!("{}", String::from_utf8(json)?);
|
||||
if let Some(error) = telemetry_error {
|
||||
anyhow::bail!("metrics-server report unavailable: {error}");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -181,6 +229,7 @@ fn run_case(client: &Client, args: &ChatCorpusArgs, prompt_case: &PromptCase) ->
|
|||
.session_group
|
||||
.clone()
|
||||
.unwrap_or_else(|| format!("{}-{}", args.session_prefix, prompt_case.index));
|
||||
let request_id = format!("{}-request-{}", args.session_prefix, prompt_case.index);
|
||||
let started = Instant::now();
|
||||
let request = request_body(args, prompt_case, &session_id);
|
||||
let response = client
|
||||
|
|
@ -188,6 +237,7 @@ fn run_case(client: &Client, args: &ChatCorpusArgs, prompt_case: &PromptCase) ->
|
|||
"{}/chat/completions",
|
||||
args.base_url.trim_end_matches('/')
|
||||
))
|
||||
.header("x-request-id", request_id)
|
||||
.json(&request)
|
||||
.send();
|
||||
match response {
|
||||
|
|
@ -609,10 +659,34 @@ fn rate(tokens: u64, wall_ms: f64) -> Option<f64> {
|
|||
(tokens > 0 && wall_ms > 0.0).then(|| tokens as f64 / (wall_ms / 1000.0))
|
||||
}
|
||||
|
||||
fn metrics_report_output_path(args: &ChatCorpusArgs, metrics_run_id: &str) -> PathBuf {
|
||||
if let Some(output) = args.metrics_report_output.as_ref() {
|
||||
return output.clone();
|
||||
}
|
||||
if let Some(output) = args.output.as_ref() {
|
||||
return sibling_path(output, "metrics-report.json");
|
||||
}
|
||||
std::env::temp_dir().join(format!("{metrics_run_id}-metrics-report.json"))
|
||||
}
|
||||
|
||||
fn generate_chat_corpus_run_id() -> String {
|
||||
let millis = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.expect("system clock before Unix epoch")
|
||||
.as_millis();
|
||||
format!("run-chat-corpus-{millis}")
|
||||
}
|
||||
|
||||
fn sibling_path(path: &Path, suffix: &str) -> PathBuf {
|
||||
let stem = path
|
||||
.file_stem()
|
||||
.and_then(|stem| stem.to_str())
|
||||
.unwrap_or("chat-corpus");
|
||||
path.with_file_name(format!("{stem}-{suffix}"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::path::PathBuf;
|
||||
|
||||
use super::*;
|
||||
|
||||
fn default_args() -> ChatCorpusArgs {
|
||||
|
|
@ -628,6 +702,10 @@ mod tests {
|
|||
include_usage: true,
|
||||
request_timeout_secs: 600,
|
||||
output: None,
|
||||
metrics_report_output: None,
|
||||
run_id: None,
|
||||
metrics_http: "http://127.0.0.1:18080".to_string(),
|
||||
metrics_run_id: None,
|
||||
session_prefix: "chat-corpus-test".to_string(),
|
||||
temperature: None,
|
||||
top_p: None,
|
||||
|
|
|
|||
|
|
@ -28,9 +28,126 @@ pub enum CommandKind {
|
|||
TokenLengths(TokenLengthsArgs),
|
||||
#[command(name = "focused-runtime")]
|
||||
FocusedRuntime(FocusedRuntimeArgs),
|
||||
Eval(EvalArgs),
|
||||
Run(RunArgs),
|
||||
}
|
||||
|
||||
#[derive(Parser)]
|
||||
pub struct EvalArgs {
|
||||
#[command(subcommand)]
|
||||
pub command: EvalCommandKind,
|
||||
}
|
||||
|
||||
#[derive(Subcommand)]
|
||||
pub enum EvalCommandKind {
|
||||
List(EvalListArgs),
|
||||
Info(EvalInfoArgs),
|
||||
Sync(EvalSyncArgs),
|
||||
Install(EvalSyncArgs),
|
||||
Doctor(EvalDoctorArgs),
|
||||
Run(EvalRunArgs),
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
#[clap(rename_all = "kebab-case")]
|
||||
pub enum EvalId {
|
||||
SpeedBench,
|
||||
TerminalBench,
|
||||
SweBenchPro,
|
||||
McpAtlas,
|
||||
}
|
||||
|
||||
impl EvalId {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::SpeedBench => "speed-bench",
|
||||
Self::TerminalBench => "terminal-bench",
|
||||
Self::SweBenchPro => "swe-bench-pro",
|
||||
Self::McpAtlas => "mcp-atlas",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
#[clap(rename_all = "kebab-case")]
|
||||
pub enum EvalPack {
|
||||
Core,
|
||||
}
|
||||
|
||||
#[derive(Parser)]
|
||||
pub struct EvalListArgs {
|
||||
#[arg(long)]
|
||||
pub cache_root: Option<PathBuf>,
|
||||
#[arg(long)]
|
||||
pub json: bool,
|
||||
}
|
||||
|
||||
#[derive(Parser)]
|
||||
pub struct EvalInfoArgs {
|
||||
pub eval: EvalId,
|
||||
#[arg(long)]
|
||||
pub cache_root: Option<PathBuf>,
|
||||
#[arg(long)]
|
||||
pub json: bool,
|
||||
}
|
||||
|
||||
#[derive(Parser)]
|
||||
pub struct EvalSyncArgs {
|
||||
#[arg(value_enum)]
|
||||
pub evals: Vec<EvalId>,
|
||||
#[arg(long, value_enum, default_value_t = EvalPack::Core)]
|
||||
pub pack: EvalPack,
|
||||
#[arg(long)]
|
||||
pub cache_root: Option<PathBuf>,
|
||||
#[arg(long)]
|
||||
pub dry_run: bool,
|
||||
}
|
||||
|
||||
#[derive(Parser)]
|
||||
pub struct EvalDoctorArgs {
|
||||
#[arg(value_enum)]
|
||||
pub evals: Vec<EvalId>,
|
||||
#[arg(long, value_enum, default_value_t = EvalPack::Core)]
|
||||
pub pack: EvalPack,
|
||||
#[arg(long)]
|
||||
pub cache_root: Option<PathBuf>,
|
||||
#[arg(long)]
|
||||
pub json: bool,
|
||||
}
|
||||
|
||||
#[derive(Parser)]
|
||||
pub struct EvalRunArgs {
|
||||
pub eval: EvalId,
|
||||
#[arg(long, default_value = "http://127.0.0.1:9337/v1")]
|
||||
pub base_url: String,
|
||||
#[arg(long, default_value = DEFAULT_LOCAL_MODEL_ID)]
|
||||
pub model: String,
|
||||
#[arg(long, default_value = "skippy-bench")]
|
||||
pub api_key: String,
|
||||
#[arg(long)]
|
||||
pub cache_root: Option<PathBuf>,
|
||||
#[arg(long)]
|
||||
pub output_dir: Option<PathBuf>,
|
||||
#[arg(long, default_value_t = 300)]
|
||||
pub timeout_secs: u64,
|
||||
#[arg(long)]
|
||||
pub harness_timeout_secs: Option<u64>,
|
||||
#[arg(
|
||||
long,
|
||||
default_value_t = 1,
|
||||
help = "Expected OpenAI endpoint generation concurrency; native harness request concurrency is kept equal to this value."
|
||||
)]
|
||||
pub endpoint_concurrency: usize,
|
||||
#[arg(long)]
|
||||
pub run_id: Option<String>,
|
||||
#[arg(long, default_value = "http://127.0.0.1:18080")]
|
||||
pub metrics_http: String,
|
||||
#[arg(long)]
|
||||
pub metrics_run_id: Option<String>,
|
||||
#[arg(long)]
|
||||
pub dry_run: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, ValueEnum)]
|
||||
pub enum FocusedRuntimeScenario {
|
||||
ColdStartup,
|
||||
|
|
@ -147,6 +264,14 @@ pub struct ChatCorpusArgs {
|
|||
pub request_timeout_secs: u64,
|
||||
#[arg(long)]
|
||||
pub output: Option<PathBuf>,
|
||||
#[arg(long)]
|
||||
pub metrics_report_output: Option<PathBuf>,
|
||||
#[arg(long)]
|
||||
pub run_id: Option<String>,
|
||||
#[arg(long, default_value = "http://127.0.0.1:18080")]
|
||||
pub metrics_http: String,
|
||||
#[arg(long)]
|
||||
pub metrics_run_id: Option<String>,
|
||||
#[arg(long, default_value = "chat-corpus-session")]
|
||||
pub session_prefix: String,
|
||||
#[arg(long)]
|
||||
|
|
@ -167,7 +292,7 @@ pub struct ChatCorpusArgs {
|
|||
pub struct RunArgs {
|
||||
#[arg(long, default_value = "target/debug/metrics-server")]
|
||||
pub metrics_server_bin: PathBuf,
|
||||
#[arg(long, default_value = "target/debug/skippy-server")]
|
||||
#[arg(long, default_value = "target/release/skippy-server")]
|
||||
pub stage_server_bin: PathBuf,
|
||||
#[arg(
|
||||
long,
|
||||
|
|
@ -300,7 +425,7 @@ pub struct RunArgs {
|
|||
pub struct LocalSingleArgs {
|
||||
#[arg(long, default_value = "target/debug/metrics-server")]
|
||||
pub metrics_server_bin: PathBuf,
|
||||
#[arg(long, default_value = "target/debug/skippy-server")]
|
||||
#[arg(long, default_value = "target/release/skippy-server")]
|
||||
pub stage_server_bin: PathBuf,
|
||||
#[arg(long)]
|
||||
pub model_path: PathBuf,
|
||||
|
|
@ -360,7 +485,7 @@ pub struct LocalSplitInprocessArgs {
|
|||
|
||||
#[derive(Parser)]
|
||||
pub struct LocalSplitBinaryArgs {
|
||||
#[arg(long, default_value = "target/debug/skippy-server")]
|
||||
#[arg(long, default_value = "target/release/skippy-server")]
|
||||
pub stage_server_bin: PathBuf,
|
||||
#[arg(long)]
|
||||
pub model_path: PathBuf,
|
||||
|
|
@ -388,7 +513,7 @@ pub struct LocalSplitBinaryArgs {
|
|||
|
||||
#[derive(Parser)]
|
||||
pub struct LocalSplitCompareArgs {
|
||||
#[arg(long, default_value = "target/debug/skippy-server")]
|
||||
#[arg(long, default_value = "target/release/skippy-server")]
|
||||
pub stage_server_bin: PathBuf,
|
||||
#[arg(long)]
|
||||
pub model_path: PathBuf,
|
||||
|
|
@ -418,7 +543,7 @@ pub struct LocalSplitCompareArgs {
|
|||
|
||||
#[derive(Parser)]
|
||||
pub struct LocalSplitChainBinaryArgs {
|
||||
#[arg(long, default_value = "target/debug/skippy-server")]
|
||||
#[arg(long, default_value = "target/release/skippy-server")]
|
||||
pub stage_server_bin: PathBuf,
|
||||
#[arg(long)]
|
||||
pub model_path: PathBuf,
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ use skippy_topology::{
|
|||
use crate::{
|
||||
cli::{DEFAULT_RUN_MAX_NEW_TOKENS, FocusedRuntimeArgs, FocusedRuntimeScenario, RunArgs},
|
||||
model_identity::model_identity_for_path,
|
||||
support::{ChildGuard, parse_wire_dtype, retry},
|
||||
support::{ChildGuard, ensure_release_skippy_server_bin, parse_wire_dtype, retry},
|
||||
};
|
||||
|
||||
struct DistributedRunOutcome {
|
||||
|
|
@ -324,6 +324,7 @@ pub fn run_distributed(args: RunArgs) -> Result<()> {
|
|||
}
|
||||
|
||||
fn run_distributed_collect(args: RunArgs) -> Result<DistributedRunOutcome> {
|
||||
ensure_release_skippy_server_bin(&args.stage_server_bin)?;
|
||||
let run_started = Instant::now();
|
||||
let hosts = parse_hosts(&args.hosts)?;
|
||||
let ranges = parse_stage_ranges(&args.splits, args.layer_end)?;
|
||||
|
|
@ -1370,14 +1371,16 @@ fn prepare_local_stage(args: &RunArgs, stage: &StageAssignment) -> Result<()> {
|
|||
stage_remote_topology_path(stage).unwrap_or_else(|_| "<unknown>".to_string())
|
||||
)
|
||||
})?;
|
||||
if args.rsync_model_artifacts
|
||||
&& let (Some(stage_model), Some(local_model)) = (
|
||||
args.stage_model.as_ref(),
|
||||
stage.local_materialized_model_path.as_ref(),
|
||||
)
|
||||
{
|
||||
materialize_stage_model_on_coordinator(stage_model, stage, local_model)?;
|
||||
if !args.rsync_model_artifacts {
|
||||
return Ok(());
|
||||
}
|
||||
let Some(stage_model) = args.stage_model.as_ref() else {
|
||||
return Ok(());
|
||||
};
|
||||
let Some(local_model) = stage.local_materialized_model_path.as_ref() else {
|
||||
return Ok(());
|
||||
};
|
||||
materialize_stage_model_on_coordinator(stage_model, stage, local_model)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
|
|||
2154
crates/skippy-bench/src/evals.rs
Normal file
2154
crates/skippy-bench/src/evals.rs
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -13,7 +13,10 @@ use serde_json::{Value, json};
|
|||
use crate::{
|
||||
cli::LocalSingleArgs,
|
||||
model_identity::model_identity_for_path,
|
||||
support::{ChildGuard, generate_run_id, retry, temp_config_path, temp_db_path},
|
||||
support::{
|
||||
ChildGuard, ensure_release_skippy_server_bin, generate_run_id, retry, temp_config_path,
|
||||
temp_db_path,
|
||||
},
|
||||
};
|
||||
|
||||
#[derive(Deserialize)]
|
||||
|
|
@ -39,6 +42,7 @@ pub fn local_single(args: LocalSingleArgs) -> Result<()> {
|
|||
if args.layer_start >= args.layer_end {
|
||||
bail!("layer_start must be less than layer_end");
|
||||
}
|
||||
ensure_release_skippy_server_bin(&args.stage_server_bin)?;
|
||||
|
||||
let client = Client::builder()
|
||||
.timeout(Duration::from_secs(15))
|
||||
|
|
|
|||
|
|
@ -23,8 +23,8 @@ use crate::{
|
|||
},
|
||||
model_identity::model_identity_for_path,
|
||||
support::{
|
||||
ChildGuard, activation_width, connect_ready, generate_run_id, parse_wire_dtype,
|
||||
temp_config_path_for,
|
||||
ChildGuard, activation_width, connect_ready, ensure_release_skippy_server_bin,
|
||||
generate_run_id, parse_wire_dtype, temp_config_path_for,
|
||||
},
|
||||
};
|
||||
|
||||
|
|
@ -273,6 +273,7 @@ fn run_binary_split(args: BinarySplitConfig) -> Result<BinarySplitResult> {
|
|||
if args.split_layer == 0 || args.split_layer >= args.layer_end {
|
||||
bail!("split_layer must be greater than zero and less than layer_end");
|
||||
}
|
||||
ensure_release_skippy_server_bin(&args.stage_server_bin)?;
|
||||
validate_local_topology_plan(
|
||||
&args.model_path,
|
||||
args.layer_end,
|
||||
|
|
@ -453,6 +454,7 @@ fn run_binary_split(args: BinarySplitConfig) -> Result<BinarySplitResult> {
|
|||
}
|
||||
|
||||
fn run_binary_chain(args: LocalSplitChainBinaryArgs) -> Result<BinaryChainResult> {
|
||||
ensure_release_skippy_server_bin(&args.stage_server_bin)?;
|
||||
if args.split_layer_1 == 0
|
||||
|| args.split_layer_1 >= args.split_layer_2
|
||||
|| args.split_layer_2 >= args.layer_end
|
||||
|
|
|
|||
|
|
@ -1,10 +1,12 @@
|
|||
mod chat_corpus;
|
||||
mod cli;
|
||||
mod distributed;
|
||||
mod evals;
|
||||
mod local_single;
|
||||
mod local_split;
|
||||
mod model_identity;
|
||||
mod support;
|
||||
mod telemetry_report;
|
||||
mod token_lengths;
|
||||
mod verify_span_local;
|
||||
|
||||
|
|
@ -15,6 +17,7 @@ use crate::{
|
|||
chat_corpus::chat_corpus,
|
||||
cli::{Cli, CommandKind},
|
||||
distributed::{focused_runtime, run_distributed},
|
||||
evals::eval_command,
|
||||
local_single::local_single,
|
||||
local_split::{
|
||||
local_split_binary, local_split_chain_binary, local_split_compare, local_split_inprocess,
|
||||
|
|
@ -34,6 +37,7 @@ fn main() -> Result<()> {
|
|||
CommandKind::ChatCorpus(args) => chat_corpus(args),
|
||||
CommandKind::TokenLengths(args) => token_lengths(args),
|
||||
CommandKind::FocusedRuntime(args) => focused_runtime(args),
|
||||
CommandKind::Eval(args) => eval_command(args),
|
||||
CommandKind::Run(args) => run_distributed(args),
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use std::{
|
||||
net::{SocketAddr, TcpStream},
|
||||
path::PathBuf,
|
||||
path::{Path, PathBuf},
|
||||
process::{Child, Command},
|
||||
thread,
|
||||
time::{Duration, SystemTime, UNIX_EPOCH},
|
||||
|
|
@ -49,6 +49,18 @@ pub fn retry(timeout_secs: u64, mut action: impl FnMut() -> Result<()>) -> Resul
|
|||
Err(last_error.unwrap_or_else(|| anyhow!("timed out")))
|
||||
}
|
||||
|
||||
pub fn ensure_release_skippy_server_bin(path: &Path) -> Result<()> {
|
||||
let path = path.to_string_lossy();
|
||||
let debug_path = path.contains("target/debug/skippy-server")
|
||||
|| path.contains("target\\debug\\skippy-server");
|
||||
if debug_path {
|
||||
bail!(
|
||||
"SkippyBench benchmark-managed skippy-server runs require a release binary; run `just release-build` and use --stage-server-bin target/release/skippy-server"
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn connect_ready(addr: SocketAddr, timeout_secs: u64) -> Result<TcpStream> {
|
||||
let attempts = timeout_secs.saturating_mul(2).max(1);
|
||||
let mut last_error = None;
|
||||
|
|
|
|||
417
crates/skippy-bench/src/telemetry_report.rs
Normal file
417
crates/skippy-bench/src/telemetry_report.rs
Normal file
|
|
@ -0,0 +1,417 @@
|
|||
use std::{collections::BTreeMap, fs, path::Path, thread, time::Duration};
|
||||
|
||||
use anyhow::{Context, Result, anyhow};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
|
||||
const OPENAI_DECODE_TOKEN_SPAN: &str = "stage.openai_decode_token";
|
||||
|
||||
#[derive(Default, Serialize)]
|
||||
pub struct BenchTelemetry {
|
||||
pub metrics_http: Option<String>,
|
||||
pub metrics_run_id: String,
|
||||
pub status: &'static str,
|
||||
pub detail: Option<String>,
|
||||
pub report_path: Option<String>,
|
||||
pub request_count: Option<u64>,
|
||||
pub span_count: Option<u64>,
|
||||
pub ttft_ms: Option<TelemetryAggregate>,
|
||||
pub fttt_ms: Option<TelemetryAggregate>,
|
||||
pub request_latency_ms: Option<TelemetryAggregate>,
|
||||
pub generation_latency_ms: Option<TelemetryAggregate>,
|
||||
}
|
||||
|
||||
#[derive(Default, Serialize)]
|
||||
pub struct TelemetryAggregate {
|
||||
pub count: usize,
|
||||
pub min: f64,
|
||||
pub mean: f64,
|
||||
pub p50: f64,
|
||||
pub p95: f64,
|
||||
pub max: f64,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct CreateRunResponse {
|
||||
run_id: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RunStatusResponse {
|
||||
status: String,
|
||||
}
|
||||
|
||||
pub fn pending(metrics_http: &str, metrics_run_id: &str) -> BenchTelemetry {
|
||||
BenchTelemetry {
|
||||
metrics_http: Some(metrics_http.to_string()),
|
||||
metrics_run_id: metrics_run_id.to_string(),
|
||||
status: "pending",
|
||||
detail: Some("telemetry collection has not run yet".to_string()),
|
||||
..BenchTelemetry::default()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn create_run(metrics_http: &str, metrics_run_id: &str, config: &Value) -> Result<()> {
|
||||
let body = create_run_body(metrics_run_id, config);
|
||||
let client = reqwest::blocking::Client::new();
|
||||
let base = metrics_http.trim_end_matches('/');
|
||||
let response = client
|
||||
.post(format!("{base}/v1/runs"))
|
||||
.json(&body)
|
||||
.send()
|
||||
.with_context(|| format!("create metrics-server run {metrics_run_id}"))?;
|
||||
if !response.status().is_success() {
|
||||
let status = response.status();
|
||||
let body = response.text().unwrap_or_default();
|
||||
if body.contains("UNIQUE constraint failed: runs.run_id") {
|
||||
return accept_existing_run(&client, base, metrics_run_id);
|
||||
}
|
||||
return Err(anyhow!(
|
||||
"metrics-server rejected run {metrics_run_id}: HTTP {status}: {body}"
|
||||
));
|
||||
}
|
||||
let response = response
|
||||
.json::<CreateRunResponse>()
|
||||
.context("decode metrics-server create-run response")?;
|
||||
if response.run_id == metrics_run_id {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(anyhow!(
|
||||
"metrics-server returned unexpected run_id {}",
|
||||
response.run_id
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn accept_existing_run(
|
||||
client: &reqwest::blocking::Client,
|
||||
metrics_http: &str,
|
||||
metrics_run_id: &str,
|
||||
) -> Result<()> {
|
||||
let status = client
|
||||
.get(format!("{metrics_http}/v1/runs/{metrics_run_id}/status"))
|
||||
.send()
|
||||
.with_context(|| format!("read metrics-server run status {metrics_run_id}"))?
|
||||
.error_for_status()
|
||||
.with_context(|| format!("metrics-server rejected status for run {metrics_run_id}"))?
|
||||
.json::<RunStatusResponse>()
|
||||
.context("decode metrics-server run status response")?;
|
||||
if status.status == "running" || status.status == "implicit" {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(anyhow!(
|
||||
"metrics-server run {metrics_run_id} already exists with status {}",
|
||||
status.status
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn create_run_body(metrics_run_id: &str, config: &Value) -> Value {
|
||||
let mut body = match config {
|
||||
Value::Object(object) => object.clone(),
|
||||
value => {
|
||||
let mut object = serde_json::Map::new();
|
||||
object.insert("config".to_string(), value.clone());
|
||||
object
|
||||
}
|
||||
};
|
||||
body.insert(
|
||||
"run_id".to_string(),
|
||||
Value::String(metrics_run_id.to_string()),
|
||||
);
|
||||
Value::Object(body)
|
||||
}
|
||||
|
||||
pub fn finalize_and_collect(
|
||||
metrics_http: &str,
|
||||
metrics_run_id: &str,
|
||||
report_path: &Path,
|
||||
) -> Result<BenchTelemetry> {
|
||||
let report = fetch_metrics_report(metrics_http, metrics_run_id, report_path)?;
|
||||
Ok(telemetry_from_report(
|
||||
metrics_http,
|
||||
metrics_run_id,
|
||||
report_path,
|
||||
report,
|
||||
))
|
||||
}
|
||||
|
||||
pub fn unavailable(
|
||||
metrics_http: &str,
|
||||
metrics_run_id: &str,
|
||||
error: &anyhow::Error,
|
||||
) -> BenchTelemetry {
|
||||
BenchTelemetry {
|
||||
metrics_http: Some(metrics_http.to_string()),
|
||||
metrics_run_id: metrics_run_id.to_string(),
|
||||
status: "unavailable",
|
||||
detail: Some(error.to_string()),
|
||||
..BenchTelemetry::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn fetch_metrics_report(
|
||||
metrics_http: &str,
|
||||
metrics_run_id: &str,
|
||||
report_path: &Path,
|
||||
) -> Result<Value> {
|
||||
let client = reqwest::blocking::Client::new();
|
||||
let base = metrics_http.trim_end_matches('/');
|
||||
client
|
||||
.post(format!("{base}/v1/runs/{metrics_run_id}/finalize"))
|
||||
.send()
|
||||
.with_context(|| format!("finalize metrics-server run {metrics_run_id}"))?
|
||||
.error_for_status()
|
||||
.with_context(|| format!("metrics-server rejected finalize for run {metrics_run_id}"))?;
|
||||
thread::sleep(Duration::from_millis(250));
|
||||
let report = client
|
||||
.get(format!("{base}/v1/runs/{metrics_run_id}/report.json"))
|
||||
.send()
|
||||
.with_context(|| format!("fetch metrics report for run {metrics_run_id}"))?
|
||||
.error_for_status()
|
||||
.with_context(|| format!("metrics report request failed for run {metrics_run_id}"))?
|
||||
.json::<Value>()
|
||||
.context("decode metrics report JSON")?;
|
||||
if let Some(parent) = report_path.parent() {
|
||||
fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
|
||||
}
|
||||
fs::write(report_path, serde_json::to_vec_pretty(&report)?)
|
||||
.with_context(|| format!("write {}", report_path.display()))?;
|
||||
Ok(report)
|
||||
}
|
||||
|
||||
fn telemetry_from_report(
|
||||
metrics_http: &str,
|
||||
metrics_run_id: &str,
|
||||
report_path: &Path,
|
||||
report: Value,
|
||||
) -> BenchTelemetry {
|
||||
let spans = report
|
||||
.get("spans")
|
||||
.and_then(Value::as_array)
|
||||
.map(Vec::as_slice)
|
||||
.unwrap_or(&[]);
|
||||
let requests = summarize_telemetry_requests(spans);
|
||||
let ttft_values = requests
|
||||
.iter()
|
||||
.filter_map(TelemetryRequestSummary::ttft_ms)
|
||||
.collect::<Vec<_>>();
|
||||
let request_latency_values = requests
|
||||
.iter()
|
||||
.filter_map(TelemetryRequestSummary::request_latency_ms)
|
||||
.collect::<Vec<_>>();
|
||||
let generation_latency_values = requests
|
||||
.iter()
|
||||
.filter_map(|summary| summary.generation_latency_ms)
|
||||
.collect::<Vec<_>>();
|
||||
let span_count = report
|
||||
.get("counts")
|
||||
.and_then(|counts| counts.get("spans"))
|
||||
.and_then(Value::as_u64)
|
||||
.or(Some(spans.len() as u64));
|
||||
let status = if spans.is_empty() {
|
||||
"no_spans"
|
||||
} else if ttft_values.is_empty() {
|
||||
"no_decode_token_spans"
|
||||
} else {
|
||||
"ok"
|
||||
};
|
||||
let detail = match status {
|
||||
"no_spans" => Some(
|
||||
"metrics report has no spans; target endpoint may not be emitting this run id"
|
||||
.to_string(),
|
||||
),
|
||||
"no_decode_token_spans" => Some(
|
||||
"metrics report has spans, but no stage.openai_decode_token spans; run endpoint with debug telemetry for server-side TTFT"
|
||||
.to_string(),
|
||||
),
|
||||
_ => None,
|
||||
};
|
||||
BenchTelemetry {
|
||||
metrics_http: Some(metrics_http.to_string()),
|
||||
metrics_run_id: metrics_run_id.to_string(),
|
||||
status,
|
||||
detail,
|
||||
report_path: Some(report_path.display().to_string()),
|
||||
request_count: Some(requests.len() as u64),
|
||||
span_count,
|
||||
ttft_ms: aggregate(&ttft_values),
|
||||
fttt_ms: aggregate(&ttft_values),
|
||||
request_latency_ms: aggregate(&request_latency_values),
|
||||
generation_latency_ms: aggregate(&generation_latency_values),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct TelemetryRequestSummary {
|
||||
first_start_unix_nanos: Option<i64>,
|
||||
last_end_unix_nanos: Option<i64>,
|
||||
first_decode_token_start_unix_nanos: Option<i64>,
|
||||
generation_latency_ms: Option<f64>,
|
||||
}
|
||||
|
||||
impl TelemetryRequestSummary {
|
||||
fn observe_span(&mut self, span: &Value) {
|
||||
let Some(start) = span.get("start_time_unix_nanos").and_then(Value::as_i64) else {
|
||||
return;
|
||||
};
|
||||
let end = span
|
||||
.get("end_time_unix_nanos")
|
||||
.and_then(Value::as_i64)
|
||||
.unwrap_or(start);
|
||||
self.first_start_unix_nanos = Some(
|
||||
self.first_start_unix_nanos
|
||||
.map(|current| current.min(start))
|
||||
.unwrap_or(start),
|
||||
);
|
||||
self.last_end_unix_nanos = Some(
|
||||
self.last_end_unix_nanos
|
||||
.map(|current| current.max(end))
|
||||
.unwrap_or(end),
|
||||
);
|
||||
if span.get("name").and_then(Value::as_str) == Some(OPENAI_DECODE_TOKEN_SPAN) {
|
||||
self.first_decode_token_start_unix_nanos = Some(
|
||||
self.first_decode_token_start_unix_nanos
|
||||
.map(|current| current.min(start))
|
||||
.unwrap_or(start),
|
||||
);
|
||||
}
|
||||
if span.get("name").and_then(Value::as_str) == Some("stage.openai_generation_summary") {
|
||||
self.generation_latency_ms = span
|
||||
.get("attributes")
|
||||
.and_then(|attrs| attrs.get("llama_stage.elapsed_ms"))
|
||||
.and_then(Value::as_f64)
|
||||
.or_else(|| Some(nanos_to_ms(end.saturating_sub(start))));
|
||||
}
|
||||
}
|
||||
|
||||
fn ttft_ms(&self) -> Option<f64> {
|
||||
Some(nanos_to_ms(
|
||||
self.first_decode_token_start_unix_nanos?
|
||||
.saturating_sub(self.first_start_unix_nanos?),
|
||||
))
|
||||
}
|
||||
|
||||
fn request_latency_ms(&self) -> Option<f64> {
|
||||
Some(nanos_to_ms(
|
||||
self.last_end_unix_nanos?
|
||||
.saturating_sub(self.first_start_unix_nanos?),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn summarize_telemetry_requests(spans: &[Value]) -> Vec<TelemetryRequestSummary> {
|
||||
let mut by_request = BTreeMap::<String, TelemetryRequestSummary>::new();
|
||||
for span in spans {
|
||||
let request_id = span.get("request_id").and_then(Value::as_str).or_else(|| {
|
||||
span.get("attributes")
|
||||
.and_then(|attrs| attrs.get("llama_stage.request_id"))
|
||||
.and_then(Value::as_str)
|
||||
});
|
||||
if let Some(request_id) = request_id {
|
||||
by_request
|
||||
.entry(request_id.to_string())
|
||||
.or_default()
|
||||
.observe_span(span);
|
||||
}
|
||||
}
|
||||
by_request.into_values().collect()
|
||||
}
|
||||
|
||||
fn aggregate(values: &[f64]) -> Option<TelemetryAggregate> {
|
||||
if values.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let mut values = values.to_vec();
|
||||
values.sort_by(f64::total_cmp);
|
||||
let sum = values.iter().sum::<f64>();
|
||||
Some(TelemetryAggregate {
|
||||
count: values.len(),
|
||||
min: values[0],
|
||||
mean: sum / values.len() as f64,
|
||||
p50: percentile(&values, 0.50),
|
||||
p95: percentile(&values, 0.95),
|
||||
max: values[values.len() - 1],
|
||||
})
|
||||
}
|
||||
|
||||
fn percentile(values: &[f64], percentile: f64) -> f64 {
|
||||
let index = ((values.len() - 1) as f64 * percentile).round() as usize;
|
||||
values[index]
|
||||
}
|
||||
|
||||
fn nanos_to_ms(nanos: i64) -> f64 {
|
||||
nanos as f64 / 1_000_000.0
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::{
|
||||
fs,
|
||||
path::PathBuf,
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use serde_json::json;
|
||||
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn telemetry_report_extracts_ttft_and_request_latency() {
|
||||
let run_dir = temp_run_dir("telemetry-metrics");
|
||||
fs::create_dir_all(run_dir.join("raw")).unwrap();
|
||||
let report = json!({
|
||||
"counts": {"spans": 3},
|
||||
"spans": [
|
||||
{
|
||||
"request_id": "req-1",
|
||||
"name": "stage.openai_tokenize",
|
||||
"start_time_unix_nanos": 1_000_000_000_i64,
|
||||
"end_time_unix_nanos": 1_010_000_000_i64,
|
||||
"attributes": {}
|
||||
},
|
||||
{
|
||||
"request_id": "req-1",
|
||||
"name": "stage.openai_decode_token",
|
||||
"start_time_unix_nanos": 1_125_000_000_i64,
|
||||
"end_time_unix_nanos": 1_130_000_000_i64,
|
||||
"attributes": {}
|
||||
},
|
||||
{
|
||||
"request_id": "req-1",
|
||||
"name": "stage.openai_generation_summary",
|
||||
"start_time_unix_nanos": 1_000_000_000_i64,
|
||||
"end_time_unix_nanos": 1_250_000_000_i64,
|
||||
"attributes": {"llama_stage.elapsed_ms": 250.0}
|
||||
}
|
||||
]
|
||||
});
|
||||
let telemetry = telemetry_from_report(
|
||||
"http://127.0.0.1:18080",
|
||||
"run-telemetry",
|
||||
&run_dir.join("raw/metrics-report.json"),
|
||||
report,
|
||||
);
|
||||
assert_eq!(telemetry.status, "ok");
|
||||
assert_eq!(telemetry.request_count, Some(1));
|
||||
assert_eq!(telemetry.span_count, Some(3));
|
||||
assert_eq!(telemetry.ttft_ms.as_ref().unwrap().mean, 125.0);
|
||||
assert_eq!(telemetry.request_latency_ms.as_ref().unwrap().mean, 250.0);
|
||||
assert_eq!(
|
||||
telemetry.generation_latency_ms.as_ref().unwrap().mean,
|
||||
250.0
|
||||
);
|
||||
let _ = fs::remove_dir_all(run_dir);
|
||||
}
|
||||
|
||||
fn temp_run_dir(label: &str) -> PathBuf {
|
||||
std::env::temp_dir().join(format!(
|
||||
"skippy-bench-{label}-{}-{}",
|
||||
std::process::id(),
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis()
|
||||
))
|
||||
}
|
||||
}
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
use std::{net::SocketAddr, path::PathBuf};
|
||||
|
||||
use crate::telemetry::TelemetryLevel;
|
||||
use clap::{Parser, Subcommand};
|
||||
use clap::{Parser, Subcommand, ValueEnum};
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(about = "Llama staged-runtime server")]
|
||||
|
|
@ -192,6 +192,20 @@ pub struct ServeOpenAiArgs {
|
|||
pub telemetry_queue_capacity: usize,
|
||||
#[arg(long, value_enum, default_value_t = TelemetryLevel::Summary)]
|
||||
pub telemetry_level: TelemetryLevel,
|
||||
#[arg(
|
||||
long = "openai-guardrails",
|
||||
value_enum,
|
||||
default_value_t = OpenAiGuardrailsCliMode::Metrics,
|
||||
help = "OpenAI compatibility guardrail mode for standalone serving: disabled, metrics, or enforce."
|
||||
)]
|
||||
pub openai_guardrails: OpenAiGuardrailsCliMode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
pub enum OpenAiGuardrailsCliMode {
|
||||
Disabled,
|
||||
Metrics,
|
||||
Enforce,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
|
@ -228,5 +242,24 @@ mod tests {
|
|||
assert_eq!(args.prefill_adaptive_start, 128);
|
||||
assert_eq!(args.prefill_adaptive_step, 128);
|
||||
assert_eq!(args.prefill_adaptive_max, 384);
|
||||
assert_eq!(args.openai_guardrails, OpenAiGuardrailsCliMode::Metrics);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serve_openai_accepts_explicit_guardrail_mode() {
|
||||
let cli = Cli::try_parse_from([
|
||||
"skippy-server",
|
||||
"serve-openai",
|
||||
"--config",
|
||||
"stage.json",
|
||||
"--openai-guardrails",
|
||||
"enforce",
|
||||
])
|
||||
.unwrap();
|
||||
|
||||
let Command::ServeOpenAi(args) = cli.command else {
|
||||
panic!("expected serve-openai command");
|
||||
};
|
||||
assert_eq!(args.openai_guardrails, OpenAiGuardrailsCliMode::Enforce);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -29,7 +29,8 @@ use openai_frontend::{
|
|||
GuardedOpenAiBackend, GuardrailMode, GuardrailPolicy, GuardrailPolicyHandle, MessageContent,
|
||||
MessageContentPart, ModelId, ModelObject, OpenAiBackend, OpenAiError, OpenAiErrorKind,
|
||||
OpenAiHookPolicy, OpenAiRequestContext, OpenAiResult, PrefillHookSignals, ReasoningEffort,
|
||||
StreamingGuardrailMode, Usage, apply_chat_hook_outcome, chat_mesh_hooks_enabled,
|
||||
RetryExhaustionMode, StreamingGuardrailMode, Usage, apply_chat_hook_outcome,
|
||||
chat_mesh_hooks_enabled,
|
||||
};
|
||||
use serde::Serialize;
|
||||
use serde_json::{Value, json};
|
||||
|
|
@ -62,7 +63,7 @@ use crate::{
|
|||
send_client_ready_hello_if_enabled, stage_output_activation_capacity,
|
||||
write_stage_message_conditioned,
|
||||
},
|
||||
cli::ServeOpenAiArgs,
|
||||
cli::{OpenAiGuardrailsCliMode, ServeOpenAiArgs},
|
||||
config::{load_json, validate_config},
|
||||
kv_integration::{KvStageIntegration, proactive_eviction_attrs, proactive_eviction_error_kind},
|
||||
runtime_state::{RuntimeSessionStats, RuntimeState, load_runtime},
|
||||
|
|
@ -186,7 +187,7 @@ pub async fn serve_openai(args: ServeOpenAiArgs) -> Result<()> {
|
|||
let decode_batcher = DecodeBatcher::new(runtime.clone(), args.generation_concurrency);
|
||||
let decode_frame_batcher =
|
||||
DecodeFrameBatcher::new(runtime.clone(), args.generation_concurrency);
|
||||
let backend = Arc::new(StageOpenAiBackend {
|
||||
let backend: Arc<dyn OpenAiBackend> = Arc::new(StageOpenAiBackend {
|
||||
runtime,
|
||||
config,
|
||||
telemetry: telemetry.clone(),
|
||||
|
|
@ -212,6 +213,8 @@ pub async fn serve_openai(args: ServeOpenAiArgs) -> Result<()> {
|
|||
decode_batcher,
|
||||
decode_frame_batcher,
|
||||
});
|
||||
let backend = OpenAiGuardrailsConfig::for_standalone_mode(args.openai_guardrails)
|
||||
.wrap_backend_with_context_limit(backend, Some(ctx_size));
|
||||
let app: Router = instrumented_openai_router(backend, telemetry.clone());
|
||||
|
||||
println!(
|
||||
|
|
@ -289,6 +292,37 @@ impl OpenAiGuardrailsConfig {
|
|||
}
|
||||
}
|
||||
|
||||
pub fn compatibility_for_skippy() -> Self {
|
||||
Self {
|
||||
target: OpenAiGuardrailsTarget::Skippy,
|
||||
policy: GuardrailPolicy {
|
||||
mode: GuardrailMode::MetricsOnly,
|
||||
apply_to_all_models: true,
|
||||
retry_exhaustion_mode: RetryExhaustionMode::PassLastText,
|
||||
..GuardrailPolicy::default()
|
||||
}
|
||||
.into(),
|
||||
compaction: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn for_standalone_mode(mode: OpenAiGuardrailsCliMode) -> Self {
|
||||
match mode {
|
||||
OpenAiGuardrailsCliMode::Disabled => Self::disabled_for_skippy(),
|
||||
OpenAiGuardrailsCliMode::Metrics => Self::compatibility_for_skippy(),
|
||||
OpenAiGuardrailsCliMode::Enforce => Self {
|
||||
target: OpenAiGuardrailsTarget::Skippy,
|
||||
policy: GuardrailPolicy {
|
||||
mode: GuardrailMode::Enforce,
|
||||
apply_to_all_models: true,
|
||||
..GuardrailPolicy::default()
|
||||
}
|
||||
.into(),
|
||||
compaction: None,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
pub fn status(&self) -> OpenAiGuardrailsStatus {
|
||||
let policy = self.policy.snapshot();
|
||||
OpenAiGuardrailsStatus {
|
||||
|
|
|
|||
|
|
@ -284,15 +284,6 @@ pub(super) fn ensure_chat_runtime_features_supported(
|
|||
"chat logprobs are parsed by openai-frontend but not yet implemented by skippy runtime",
|
||||
));
|
||||
}
|
||||
if request
|
||||
.response_format
|
||||
.as_ref()
|
||||
.is_some_and(requires_structured_output)
|
||||
{
|
||||
return Err(OpenAiError::unsupported(
|
||||
"structured output is parsed by openai-frontend but not yet implemented by skippy runtime",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -311,14 +302,6 @@ pub(super) fn has_requested_tools(value: &Value) -> bool {
|
|||
!matches!(value, Value::Array(items) if items.is_empty())
|
||||
}
|
||||
|
||||
pub(super) fn requires_structured_output(value: &Value) -> bool {
|
||||
value
|
||||
.as_object()
|
||||
.and_then(|object| object.get("type"))
|
||||
.and_then(Value::as_str)
|
||||
.is_some_and(|format_type| format_type != "text")
|
||||
}
|
||||
|
||||
pub(super) fn ensure_extra_generation_fields_absent(
|
||||
extra: &std::collections::BTreeMap<String, serde_json::Value>,
|
||||
) -> OpenAiResult<()> {
|
||||
|
|
|
|||
|
|
@ -651,7 +651,7 @@ fn chat_runtime_feature_guard_allows_noop_parity_fields() {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn chat_runtime_feature_guard_rejects_structured_output() {
|
||||
fn chat_runtime_feature_guard_allows_structured_output_for_guardrails() {
|
||||
let request: ChatCompletionRequest = serde_json::from_value(json!({
|
||||
"model": "test",
|
||||
"messages": [{"role": "user", "content": "hi"}],
|
||||
|
|
@ -662,11 +662,7 @@ fn chat_runtime_feature_guard_rejects_structured_output() {
|
|||
}))
|
||||
.unwrap();
|
||||
|
||||
let error = ensure_chat_runtime_features_supported(&request).unwrap_err();
|
||||
assert_eq!(
|
||||
unsupported_code(error),
|
||||
Some("unsupported_model_feature".to_string())
|
||||
);
|
||||
ensure_chat_runtime_features_supported(&request).unwrap();
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
|
|
@ -780,6 +776,29 @@ async fn guarded_structured_output_is_not_rejected_by_runtime_feature_guard() {
|
|||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standalone_guardrail_modes_have_expected_policies() {
|
||||
let metrics =
|
||||
OpenAiGuardrailsConfig::for_standalone_mode(crate::cli::OpenAiGuardrailsCliMode::Metrics)
|
||||
.status();
|
||||
assert_eq!(metrics.mode, "metrics");
|
||||
assert_eq!(metrics.retry_exhaustion, "pass_last_text");
|
||||
assert_eq!(metrics.small_model_policy, "all");
|
||||
|
||||
let enforce =
|
||||
OpenAiGuardrailsConfig::for_standalone_mode(crate::cli::OpenAiGuardrailsCliMode::Enforce)
|
||||
.status();
|
||||
assert_eq!(enforce.mode, "enforce");
|
||||
assert_eq!(enforce.retry_exhaustion, "error");
|
||||
assert_eq!(enforce.small_model_policy, "all");
|
||||
|
||||
let disabled =
|
||||
OpenAiGuardrailsConfig::for_standalone_mode(crate::cli::OpenAiGuardrailsCliMode::Disabled)
|
||||
.status();
|
||||
assert_eq!(disabled.mode, "disabled");
|
||||
assert_eq!(disabled.small_model_policy, "small_models_only");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn compaction_wraps_skippy_backend_even_when_guardrails_are_disabled() {
|
||||
let backend = Arc::new(StructuredGuardrailRecordingBackend::default());
|
||||
|
|
@ -2227,7 +2246,7 @@ fn explicit_completion_request_values_override_request_defaults() {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn request_defaults_do_not_make_structured_output_or_logprobs_executable() {
|
||||
fn request_defaults_do_not_make_logprobs_executable() {
|
||||
let mut request: ChatCompletionRequest = serde_json::from_value(json!({
|
||||
"model": "test",
|
||||
"messages": [{"role": "user", "content": "hi"}],
|
||||
|
|
|
|||
|
|
@ -644,12 +644,12 @@ openai_structured_status="$(
|
|||
-H 'content-type: application/json' \
|
||||
-d "$openai_structured_request"
|
||||
)"
|
||||
if [[ "$openai_structured_status" != "400" ]]; then
|
||||
echo "expected structured-output request to return HTTP 400 until backend support lands, got ${openai_structured_status}" >&2
|
||||
if [[ "$openai_structured_status" != "200" ]]; then
|
||||
echo "expected structured-output request to be accepted by the OpenAI compatibility layer, got ${openai_structured_status}" >&2
|
||||
cat "$openai_structured_response" >&2 || true
|
||||
exit 1
|
||||
fi
|
||||
assert_json "$openai_structured_response" '.error.code == "unsupported_model_feature"'
|
||||
assert_json "$openai_structured_response" '.choices[0].message.role == "assistant"'
|
||||
|
||||
cleanup
|
||||
SERVER_PID=""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue