diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index 59498a8d7260..a3cb665e2dcc 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -4430,6 +4430,7 @@ async fn responses( text: request.inner.text.clone(), service_tier: request.inner.service_tier.clone(), include: request.inner.include.clone(), + top_logprobs: request.inner.top_logprobs, truncation: request.inner.truncation, // Upstream `CreateResponse` doesn't carry these yet; plumbed through so // the response serializer can default to 0.0 without hardcoding at the diff --git a/lib/llm/src/protocols/openai/chat_completions.rs b/lib/llm/src/protocols/openai/chat_completions.rs index 8197c423e446..4940830ea399 100644 --- a/lib/llm/src/protocols/openai/chat_completions.rs +++ b/lib/llm/src/protocols/openai/chat_completions.rs @@ -310,6 +310,21 @@ pub enum ResponseOutputEvent { Boundary(MessageBoundary), Text(String), ToolCall(String), + /// The next `n` entries of choice 0's aggregated `logprobs.content` arrived at + /// this point of the stream on chunks with visible text or no content (EOS), + /// so a Responses projection can give each message the scores generated while + /// it was open. + Scores(usize), + /// The next `n` entries arrived on reasoning or tool-call chunks. They are not + /// output text, so a Responses projection skips them. + HiddenScores(usize), +} + +impl ResponseOutputEvent { + /// Score bookkeeping that never changes the projected output items. + pub fn is_score(&self) -> bool { + matches!(self, Self::Scores(_) | Self::HiddenScores(_)) + } } /// A response structure for unary chat completion responses, embedding OpenAI's diff --git a/lib/llm/src/protocols/openai/chat_completions/aggregator.rs b/lib/llm/src/protocols/openai/chat_completions/aggregator.rs index 445aa86b279b..956bd34e4c1c 100644 --- a/lib/llm/src/protocols/openai/chat_completions/aggregator.rs +++ b/lib/llm/src/protocols/openai/chat_completions/aggregator.rs @@ -8,7 +8,8 @@ use std::sync::Arc; use dynamo_parsers::tool_calling::try_tool_call_parse_aggregate_finalize; use super::{ - NvCreateChatCompletionResponse, NvCreateChatCompletionStreamResponse, ResponseOutputEvent, + MessageBoundary, NvCreateChatCompletionResponse, NvCreateChatCompletionStreamResponse, + ResponseOutputEvent, }; use crate::protocols::{ Annotated, @@ -329,6 +330,22 @@ impl Default for DeltaAggregator { } } +/// The last event that shapes output items, looking past score bookkeeping, so a +/// reasoning or text run merges the same way whether or not scores were requested. +fn last_content_event(events: &mut [ResponseOutputEvent]) -> Option<&mut ResponseOutputEvent> { + events.iter_mut().rev().find(|event| !event.is_score()) +} + +/// Record `count` more scores, folding into a trailing event of the same kind. +fn push_score_event(events: &mut Vec, count: usize, hidden: bool) { + match (events.last_mut(), hidden) { + (Some(ResponseOutputEvent::Scores(total)), false) + | (Some(ResponseOutputEvent::HiddenScores(total)), true) => *total += count, + (_, false) => events.push(ResponseOutputEvent::Scores(count)), + (_, true) => events.push(ResponseOutputEvent::HiddenScores(count)), + } +} + /// Merge an incoming chunk into the per-index accumulator. /// /// #8640: the prior implementation required `id`, `name`, and `arguments` @@ -535,20 +552,27 @@ impl DeltaAggregator { merge_response_nvext(&mut aggregator.nvext, delta.nvext); - if let Some(boundary) = delta.message_boundary { + // A `Start` boundary opens the message before this chunk's events; an + // `End` boundary closes it after them, so scores carried on the + // ending chunk still belong to the message that ends. + let boundary = delta.message_boundary; + if boundary.is_some() { aggregator.has_message_phases = true; - aggregator.response_events.push(ResponseOutputEvent::Boundary(boundary)); + } + if let Some(MessageBoundary::Start(_)) = &boundary { + aggregator.response_events.push(ResponseOutputEvent::Boundary(boundary.clone().unwrap())); } for choice in delta.inner.choices.iter().filter(|choice| choice.index == 0) { if let Some(reasoning) = &choice.delta.reasoning_content { - if let Some(ResponseOutputEvent::Reasoning(previous)) = aggregator.response_events.last_mut() { + if let Some(ResponseOutputEvent::Reasoning(previous)) = last_content_event(&mut aggregator.response_events) { previous.push_str(reasoning); } else { aggregator.response_events.push(ResponseOutputEvent::Reasoning(reasoning.clone())); } } + let has_text = matches!(&choice.delta.content, Some(ChatCompletionMessageContent::Text(text)) if !text.is_empty()); if let Some(ChatCompletionMessageContent::Text(text)) = &choice.delta.content { - if let Some(ResponseOutputEvent::Text(previous)) = aggregator.response_events.last_mut() { + if let Some(ResponseOutputEvent::Text(previous)) = last_content_event(&mut aggregator.response_events) { previous.push_str(text); } else { aggregator.response_events.push(ResponseOutputEvent::Text(text.clone())); @@ -559,6 +583,20 @@ impl DeltaAggregator { aggregator.response_events.push(ResponseOutputEvent::ToolCall(id.clone())); } } + // Record only the count: the entries themselves are appended to + // the choice's aggregated logprobs below, in the same order. + // Scores on reasoning or tool-call chunks are not output text. + if let Some(content) = choice.logprobs.as_ref().and_then(|lp| lp.content.as_ref()) + && !content.is_empty() + { + let hidden = !has_text + && (choice.delta.reasoning_content.as_deref().is_some_and(|text| !text.is_empty()) + || choice.delta.tool_calls.as_ref().is_some_and(|calls| !calls.is_empty())); + push_score_event(&mut aggregator.response_events, content.len(), hidden); + } + } + if let Some(MessageBoundary::End) = boundary { + aggregator.response_events.push(ResponseOutputEvent::Boundary(MessageBoundary::End)); } if let Some(prompt_logprobs) = delta.prompt_logprobs { aggregator.prompt_logprobs = Some(prompt_logprobs); @@ -955,10 +993,16 @@ impl DeltaAggregator { // Construct the final response object. let response = NvCreateChatCompletionResponse { + // Without message phases only the score bookkeeping is kept, so a + // Responses projection can still drop reasoning and tool-call scores. response_events: if aggregator.has_message_phases { aggregator.response_events } else { - Vec::new() + aggregator + .response_events + .into_iter() + .filter(ResponseOutputEvent::is_score) + .collect() }, inner: dynamo_protocols::types::CreateChatCompletionResponse { id: aggregator.id, @@ -1091,6 +1135,134 @@ mod tests { use crate::protocols::openai::token_to_utf8_bytes; use futures::stream; + /// Score counts are recorded between the message boundaries in arrival order, so a + /// Responses projection can give each message the scores generated while it was open. + #[tokio::test] + async fn phase_boundaries_record_score_counts_in_order() { + use crate::protocols::openai::chat_completions::MessageBoundary; + use dynamo_protocols::types::responses::MessagePhase; + + let boundary = |mut delta: Annotated, + boundary: MessageBoundary| { + delta.data.as_mut().unwrap().message_boundary = Some(boundary); + delta + }; + let reasoning = |text: &str, logprob: f32| { + let mut delta = create_test_delta(0, "", None, None, Some(logprob), None); + let choice = &mut delta.data.as_mut().unwrap().inner.choices[0]; + choice.delta.content = None; + choice.delta.reasoning_content = Some(text.to_string()); + delta + }; + let deltas = vec![ + // Two scored reasoning chunks still merge into one reasoning event. + reasoning("Let me ", -0.05), + reasoning("think.", -0.06), + boundary( + create_test_delta( + 0, + "Checking.", + Some(dynamo_protocols::types::Role::Assistant), + None, + Some(-0.1), + None, + ), + MessageBoundary::Start(MessagePhase::Commentary), + ), + // The ending chunk carries the message's own scores. + boundary( + create_test_delta(0, "", None, None, Some(-0.15), None), + MessageBoundary::End, + ), + boundary( + create_test_delta(0, "Done.", None, None, Some(-0.2), None), + MessageBoundary::Start(MessagePhase::FinalAnswer), + ), + boundary( + create_test_delta(0, "", None, None, None, None), + MessageBoundary::End, + ), + create_test_delta( + 0, + "", + None, + Some(dynamo_protocols::types::FinishReason::Stop), + Some(-0.3), + None, + ), + ]; + + let resp = DeltaAggregator::apply(stream::iter(deltas), ParsingOptions::default()) + .await + .unwrap(); + let events: Vec<_> = resp + .response_events + .into_iter() + .filter(|event| !matches!(event, ResponseOutputEvent::Text(text) if text.is_empty())) + .collect(); + assert_eq!( + events, + vec![ + ResponseOutputEvent::Reasoning("Let me think.".into()), + ResponseOutputEvent::HiddenScores(2), + ResponseOutputEvent::Boundary(MessageBoundary::Start(MessagePhase::Commentary)), + ResponseOutputEvent::Text("Checking.".into()), + ResponseOutputEvent::Scores(2), + ResponseOutputEvent::Boundary(MessageBoundary::End), + ResponseOutputEvent::Boundary(MessageBoundary::Start(MessagePhase::FinalAnswer)), + ResponseOutputEvent::Text("Done.".into()), + ResponseOutputEvent::Scores(1), + ResponseOutputEvent::Boundary(MessageBoundary::End), + ResponseOutputEvent::Scores(1), + ] + ); + let scored = resp.inner.choices[0] + .logprobs + .as_ref() + .and_then(|lp| lp.content.as_ref()) + .map(Vec::len); + assert_eq!(scored, Some(6)); + } + + /// Without message phases the response still carries the score bookkeeping, and + /// nothing else, so hidden scores can be dropped from a Responses projection. + #[tokio::test] + async fn plain_responses_keep_only_score_events() { + let mut thinking = create_test_delta(0, "", None, None, Some(-0.05), None); + let choice = &mut thinking.data.as_mut().unwrap().inner.choices[0]; + choice.delta.content = None; + choice.delta.reasoning_content = Some("Thinking.".to_string()); + let deltas = vec![ + thinking, + create_test_delta( + 0, + "Hello", + Some(dynamo_protocols::types::Role::Assistant), + None, + Some(-0.1), + None, + ), + create_test_delta( + 0, + "", + None, + Some(dynamo_protocols::types::FinishReason::Stop), + Some(-0.3), + None, + ), + ]; + let resp = DeltaAggregator::apply(stream::iter(deltas), ParsingOptions::default()) + .await + .unwrap(); + assert_eq!( + resp.response_events, + vec![ + ResponseOutputEvent::HiddenScores(1), + ResponseOutputEvent::Scores(2), + ] + ); + } + #[tokio::test] async fn v2_batch_parser_uses_request_tool_schema() { if crate::test_utils::run_isolated( diff --git a/lib/llm/src/protocols/openai/responses/mod.rs b/lib/llm/src/protocols/openai/responses/mod.rs index 4889c397f52d..c580c66027db 100644 --- a/lib/llm/src/protocols/openai/responses/mod.rs +++ b/lib/llm/src/protocols/openai/responses/mod.rs @@ -20,13 +20,14 @@ use std::collections::{HashMap, HashSet}; use dynamo_protocols::types::responses::{ AssistantRole, FunctionCallOutput, FunctionToolCall, IncludeEnum, IncompleteDetails, InputContent, InputImageContent, InputItem, InputOutputMessageContent, InputParam, InputRole, - InputTokenDetails, Instructions, Item, MessageItem, MessagePhase, NamespaceToolParamTool, - OutputItem, OutputMessage, OutputMessageContent, OutputStatus, OutputTextContent, - OutputTokenDetails, PromptCacheRetention, Reasoning, ReasoningItem, ReasoningItemContent, - ReasoningTextContent, RefusalContent, RequestTool, RequestToolChoiceParam, Response, - ResponseTextParam, ResponseUsage, Role as ResponseRole, ServiceTierResponses, Status, - SummaryPart, TextResponseFormatConfiguration, Tool, ToolChoiceAllowed, ToolChoiceAllowedMode, - ToolChoiceOptions, ToolSearchExecutionType, Truncation, + InputTokenDetails, Instructions, Item, LogProb, MessageItem, MessagePhase, + NamespaceToolParamTool, OutputItem, OutputMessage, OutputMessageContent, OutputStatus, + OutputTextContent, OutputTokenDetails, PromptCacheRetention, Reasoning, ReasoningItem, + ReasoningItemContent, ReasoningTextContent, RefusalContent, RequestTool, + RequestToolChoiceParam, Response, ResponseLogProb, ResponseTextParam, ResponseTopLobProb, + ResponseUsage, Role as ResponseRole, ServiceTierResponses, Status, SummaryPart, + TextResponseFormatConfiguration, Tool, ToolChoiceAllowed, ToolChoiceAllowedMode, + ToolChoiceOptions, ToolSearchExecutionType, TopLogProb, Truncation, }; use dynamo_protocols::types::{ ChatCompletionMessageToolCall, ChatCompletionNamedToolChoice, @@ -37,10 +38,10 @@ use dynamo_protocols::types::{ ChatCompletionRequestToolMessage, ChatCompletionRequestToolMessageContent, ChatCompletionRequestToolMessageContentPart, ChatCompletionRequestUserMessage, ChatCompletionRequestUserMessageContent, ChatCompletionRequestUserMessageContentPart, - ChatCompletionTool, ChatCompletionToolChoiceOption, ChatCompletionToolType, - CreateChatCompletionRequest, FinishReason, FunctionName, FunctionObject, FunctionType, - ImageUrl, ReasoningContent, ReasoningEffort as ChatReasoningEffort, ResponseFormat, - ServiceTier as ChatServiceTier, + ChatCompletionTokenLogprob, ChatCompletionTool, ChatCompletionToolChoiceOption, + ChatCompletionToolType, CreateChatCompletionRequest, FinishReason, FunctionName, + FunctionObject, FunctionType, ImageUrl, ReasoningContent, + ReasoningEffort as ChatReasoningEffort, ResponseFormat, ServiceTier as ChatServiceTier, }; use dynamo_runtime::protocols::annotated::AnnotationsProvider; use serde::{Deserialize, Serialize}; @@ -1250,7 +1251,11 @@ impl TryFrom for NvCreateChatCompletionRequest { history_phases.clear(); } - let top_logprobs = convert_top_logprobs(resp.inner.top_logprobs); + // Scores need the include selector plus an explicit count (chat defaults to 1). + // A bare `top_logprobs` is only echoed: chat rejects it without `logprobs: true`. + let logprobs_requested = include_requests_logprobs(resp.inner.include.as_deref()); + let top_logprobs = + logprobs_requested.then(|| convert_top_logprobs(resp.inner.top_logprobs).unwrap_or(0)); let (tools, tool_choice) = convert_tools_and_choice( (!inner_tools.is_empty()).then_some(inner_tools.as_slice()), @@ -1342,6 +1347,7 @@ impl TryFrom for NvCreateChatCompletionRequest { max_completion_tokens: resp.inner.max_output_tokens, store: resp.inner.store, parallel_tool_calls: resp.inner.parallel_tool_calls, + logprobs: logprobs_requested.then_some(true), top_logprobs, metadata: resp .inner @@ -1368,6 +1374,57 @@ impl TryFrom for NvCreateChatCompletionRequest { } } +/// Whether `include` asks for token log probabilities on output text. +pub(crate) fn include_requests_logprobs(include: Option<&[IncludeEnum]>) -> bool { + include.is_some_and(|include| include.contains(&IncludeEnum::MessageOutputTextLogprobs)) +} + +/// Project chat token records onto Responses `output_text` logprob entries. +/// +/// Responses has one logprob slot, on the text part. Chat records every +/// generated token there in generation order, including reasoning, tool-call +/// and control tokens that the parsed text does not show, so the entries need +/// not line up with the visible text. +pub(crate) fn chat_logprobs_to_output_text(entries: &[ChatCompletionTokenLogprob]) -> Vec { + entries + .iter() + .map(|entry| LogProb { + token: entry.token.clone(), + logprob: f64::from(entry.logprob), + bytes: entry.bytes.clone().unwrap_or_default(), + top_logprobs: entry + .top_logprobs + .iter() + .map(|top| TopLogProb { + token: top.token.clone(), + logprob: f64::from(top.logprob), + bytes: top.bytes.clone().unwrap_or_default(), + }) + .collect(), + }) + .collect() +} + +/// Project `output_text` logprob entries onto the stream event form, which +/// carries no `bytes`. +pub(crate) fn output_text_logprobs_to_events(entries: &[LogProb]) -> Vec { + entries + .iter() + .map(|entry| ResponseLogProb { + token: entry.token.clone(), + logprob: entry.logprob, + top_logprobs: entry + .top_logprobs + .iter() + .map(|top| ResponseTopLobProb { + token: top.token.clone(), + logprob: top.logprob, + }) + .collect(), + }) + .collect() +} + fn convert_top_logprobs(input: Option) -> Option { input.map(|x| x.min(20)) } @@ -1399,6 +1456,8 @@ pub struct ResponseParams { pub text: Option, pub service_tier: Option, pub include: Option>, + /// Requested top-k per token, echoed on the response object. + pub top_logprobs: Option, pub truncation: Option, /// OpenResponses spec requires these fields on the response body. Upstream /// `CreateResponse` doesn't model them on the request yet, so for now they @@ -1432,6 +1491,17 @@ pub(crate) fn responses_incomplete_reason( } impl ResponseParams { + /// Whether `include` asks for token log probabilities on output text. + pub(crate) fn logprobs_requested(&self) -> bool { + include_requests_logprobs(self.include.as_deref()) + } + + /// The `top_logprobs` value the response object reports: the count the + /// backend was asked for, so a request above the supported cap sees the cap. + pub(crate) fn echoed_top_logprobs(&self) -> u8 { + convert_top_logprobs(self.top_logprobs).unwrap_or(0) + } + fn reasoning_summary_requested(&self) -> bool { self.reasoning .as_ref() @@ -1545,9 +1615,42 @@ pub fn chat_completion_to_response( let choice = chat_resp.choices.into_iter().next(); let mut output = Vec::new(); let mut incomplete_reason = None; + // Every token record the backend scored, in generation order. Responses + // carries them on the text part, so hold them until the output exists. + let mut text_logprobs = Vec::new(); if let Some(choice) = choice { incomplete_reason = responses_incomplete_reason(choice.finish_reason); + if params.logprobs_requested() + && let Some(entries) = choice + .logprobs + .as_ref() + .and_then(|lp| lp.content.as_deref()) + { + text_logprobs = chat_logprobs_to_output_text(entries); + // Scores on reasoning or tool-call chunks are not output text. The + // phase path drops them while placing scores; a response without + // phases carries only the score bookkeeping and is filtered here. + if !response_events.is_empty() + && response_events + .iter() + .all(super::chat_completions::ResponseOutputEvent::is_score) + { + let mut record = std::mem::take(&mut text_logprobs).into_iter(); + for event in &response_events { + match event { + super::chat_completions::ResponseOutputEvent::Scores(count) => { + text_logprobs.extend(record.by_ref().take(*count)); + } + super::chat_completions::ResponseOutputEvent::HiddenScores(count) => { + record.by_ref().take(*count).for_each(drop); + } + _ => {} + } + } + text_logprobs.extend(record); + } + } // Reasoning precedes tool calls so output order matches the decoded turn. if let Some(reasoning_text) = choice.message.reasoning_content @@ -1624,7 +1727,7 @@ pub fn chat_completion_to_response( })); } - if !response_events.is_empty() { + if response_events.iter().any(|event| !event.is_score()) { // Batch parsing can supply boundaries without ordered reasoning // events. Preserve its aggregate reasoning in that case; explicit // ordered events are authoritative and must not duplicate it. @@ -1638,11 +1741,18 @@ pub fn chat_completion_to_response( let mut phase = None; let mut text = String::new(); let mut open = false; + // Scores are consumed from the aggregated record in event order, so each + // message carries the scores generated while it was open; scores from + // reasoning or tool calls between messages ride on the next message. + let mut scores = std::mem::take(&mut text_logprobs).into_iter(); + let mut pending_scores: Vec = Vec::new(); + let mut saw_scores = false; let flush = |output: &mut Vec, text: &mut String, phase: Option, status: OutputStatus, - open: &mut bool| { + open: &mut bool, + pending_scores: &mut Vec| { if *open || !text.is_empty() { let OutputItem::Message(mut message) = make_text_message( format!("msg_{}", Uuid::new_v4().simple()), @@ -1652,6 +1762,11 @@ pub fn chat_completion_to_response( }; message.phase = phase; message.status = status; + if let Some(OutputMessageContent::OutputText(part)) = + message.content.first_mut() + { + part.logprobs = Some(std::mem::take(pending_scores)); + } output.push(OutputItem::Message(message)); *open = false; } @@ -1659,6 +1774,14 @@ pub fn chat_completion_to_response( use super::chat_completions::{MessageBoundary, ResponseOutputEvent}; for event in response_events { match event { + ResponseOutputEvent::Scores(count) => { + saw_scores = true; + pending_scores.extend(scores.by_ref().take(count)); + } + ResponseOutputEvent::HiddenScores(count) => { + saw_scores = true; + scores.by_ref().take(count).for_each(drop); + } ResponseOutputEvent::Boundary(MessageBoundary::Start(next)) => { flush( &mut output, @@ -1666,6 +1789,7 @@ pub fn chat_completion_to_response( phase, OutputStatus::Completed, &mut open, + &mut pending_scores, ); phase = Some(next); open = true; @@ -1677,6 +1801,7 @@ pub fn chat_completion_to_response( phase, OutputStatus::Completed, &mut open, + &mut pending_scores, ); phase = None; } @@ -1687,6 +1812,7 @@ pub fn chat_completion_to_response( phase, OutputStatus::Completed, &mut open, + &mut pending_scores, ); if params.reasoning_summary_requested() && !reasoning.is_empty() { output.push(OutputItem::Reasoning(ReasoningItem { @@ -1708,6 +1834,7 @@ pub fn chat_completion_to_response( phase, OutputStatus::Completed, &mut open, + &mut pending_scores, ); if let Some(call) = replay_calls.iter().find(|call| call.id == id) { output.push(params.tool_output( @@ -1731,7 +1858,31 @@ pub fn chat_completion_to_response( OutputStatus::Completed }, &mut open, + &mut pending_scores, ); + // Scores after the last message (the hidden EOS) complete the record of + // the last text part. A record without score events (batch parsing) + // stays whole for the first text part below. + pending_scores.extend(scores); + if saw_scores + && let Some(part) = output.iter_mut().rev().find_map(|item| match item { + OutputItem::Message(message) => { + message + .content + .iter_mut() + .find_map(|content| match content { + OutputMessageContent::OutputText(part) => Some(part), + _ => None, + }) + } + _ => None, + }) + { + part.logprobs + .get_or_insert_default() + .extend(std::mem::take(&mut pending_scores)); + } + text_logprobs = pending_scores; // Refusals have no parser boundary event. Retain them on the final // visible message rather than dropping them during reconstruction. if let Some(refusal) = refusal { @@ -1766,19 +1917,20 @@ pub fn chat_completion_to_response( output.push(make_text_message(message_id, String::new())); } - // Apply `include` filtering: strip logprobs from output text unless - // the caller explicitly requested them via `message.output_text.logprobs`. - let keep_logprobs = params - .include - .as_ref() - .is_some_and(|inc| inc.contains(&IncludeEnum::MessageOutputTextLogprobs)); + // Scores not yet placed by the ordered events go on the first text part that + // has none; every other text part reports an empty array, which the schema + // requires. Without `include`, the record is empty and every part is `[]`. + let mut text_logprobs = Some(text_logprobs); for item in &mut output { if let OutputItem::Message(msg) = item { for content in &mut msg.content { if let OutputMessageContent::OutputText(text) = content - && (!keep_logprobs || text.logprobs.is_none()) + && text + .logprobs + .as_ref() + .is_none_or(|scored| scored.is_empty()) { - text.logprobs = Some(Vec::new()); + text.logprobs = Some(text_logprobs.take().unwrap_or_default()); } } } @@ -1884,7 +2036,7 @@ pub fn chat_completion_to_response( .clone() .unwrap_or(ServiceTierResponses::Auto), ), - top_logprobs: Some(0), + top_logprobs: Some(params.echoed_top_logprobs()), usage: chat_resp.usage.map(|u| ResponseUsage { input_tokens: u.prompt_tokens, input_tokens_details: InputTokenDetails { @@ -2012,7 +2164,8 @@ mod tests { assert_eq!(nv_req.inner.temperature, Some(0.5)); assert_eq!(nv_req.inner.top_p, Some(0.9)); assert_eq!(nv_req.inner.max_completion_tokens, Some(1024)); - assert_eq!(nv_req.inner.top_logprobs, Some(15)); + // Scores were not requested, so the count does not reach the backend. + assert_eq!(nv_req.inner.top_logprobs, None); assert_eq!(nv_req.inner.stream, Some(true)); let messages = &nv_req.inner.messages; @@ -2028,6 +2181,203 @@ mod tests { } } + /// `include: message.output_text.logprobs` is what asks for scores; it turns on + /// chat `logprobs` and defaults `top_logprobs` to 0 so the chat default of 1 does + /// not apply. `top_logprobs` alone requests nothing and is not forwarded, because + /// chat rejects a count without `logprobs: true`. + #[test] + fn include_logprobs_enables_chat_logprobs() { + let nv_req: NvCreateChatCompletionRequest = + make_response_with_input("hi").try_into().unwrap(); + assert_eq!(nv_req.inner.logprobs, None); + assert_eq!(nv_req.inner.top_logprobs, None); + + let mut request = make_response_with_input("hi"); + request.inner.include = Some(vec![IncludeEnum::MessageOutputTextLogprobs]); + let nv_req: NvCreateChatCompletionRequest = request.try_into().unwrap(); + assert_eq!(nv_req.inner.logprobs, Some(true)); + assert_eq!(nv_req.inner.top_logprobs, Some(15)); + + let mut request = make_response_with_input("hi"); + request.inner.include = Some(vec![IncludeEnum::MessageOutputTextLogprobs]); + request.inner.top_logprobs = None; + let nv_req: NvCreateChatCompletionRequest = request.try_into().unwrap(); + assert_eq!(nv_req.inner.logprobs, Some(true)); + assert_eq!(nv_req.inner.top_logprobs, Some(0)); + } + + /// Aggregated chat scores land on the output text part, with bytes and + /// alternatives, and the response echoes the requested `top_logprobs`. + #[test] + fn included_logprobs_map_onto_output_text() { + let mut chat_resp = make_chat_resp_with_text("hi"); + chat_resp.inner.choices[0].logprobs = Some( + serde_json::from_value(serde_json::json!({ + "content": [ + {"token": "h", "logprob": -0.25, "bytes": [104], + "top_logprobs": [{"token": "h", "logprob": -0.25, "bytes": [104]}, + {"token": "H", "logprob": -1.5, "bytes": [72]}]}, + {"token": "i", "logprob": -0.5, "bytes": null, "top_logprobs": []}, + {"token": "", "logprob": -0.75, "bytes": [], "top_logprobs": []} + ], + "refusal": null + })) + .unwrap(), + ); + let params = ResponseParams { + include: Some(vec![IncludeEnum::MessageOutputTextLogprobs]), + top_logprobs: Some(2), + ..Default::default() + }; + + let resp = chat_completion_to_response(chat_resp.clone(), ¶ms, None).unwrap(); + assert_eq!(resp.inner.top_logprobs, Some(2)); + let capped = ResponseParams { + top_logprobs: Some(21), + ..params.clone() + }; + assert_eq!( + capped.echoed_top_logprobs(), + 20, + "echo the count sent to chat" + ); + let OutputItem::Message(message) = &resp.inner.output[0] else { + panic!("expected message output"); + }; + let OutputMessageContent::OutputText(text) = &message.content[0] else { + panic!("expected text content"); + }; + let logprobs = text.logprobs.as_ref().unwrap(); + let tokens: Vec<_> = logprobs.iter().map(|entry| entry.token.as_str()).collect(); + assert_eq!(tokens, ["h", "i", ""]); + assert_eq!(logprobs[0].logprob, -0.25); + assert_eq!(logprobs[0].bytes, vec![104]); + assert_eq!(logprobs[0].top_logprobs.len(), 2); + assert_eq!(logprobs[0].top_logprobs[1].token, "H"); + assert!( + logprobs[1].bytes.is_empty(), + "null bytes become an empty array" + ); + + let resp = + chat_completion_to_response(chat_resp, &ResponseParams::default(), None).unwrap(); + assert_eq!(resp.inner.top_logprobs, Some(0)); + let OutputItem::Message(message) = &resp.inner.output[0] else { + panic!("expected message output"); + }; + let OutputMessageContent::OutputText(text) = &message.content[0] else { + panic!("expected text content"); + }; + assert_eq!(text.logprobs.as_deref(), Some(&[][..])); + } + + /// With ordered response events, each message gets the scores generated while + /// it was open, and the trailing EOS score completes the last message. + #[test] + fn included_logprobs_follow_message_phases() { + use crate::protocols::openai::chat_completions::{MessageBoundary, ResponseOutputEvent}; + use dynamo_protocols::types::responses::MessagePhase; + + let mut chat_resp = make_chat_resp_with_text("Checking.Done."); + chat_resp.inner.choices[0].logprobs = Some( + serde_json::from_value(serde_json::json!({ + "content": [ + {"token": "think", "logprob": -0.1, "bytes": [116], "top_logprobs": []}, + {"token": "Checking.", "logprob": -0.25, "bytes": [67], "top_logprobs": []}, + {"token": "Done.", "logprob": -0.5, "bytes": [68], "top_logprobs": []}, + {"token": "", "logprob": -0.75, "bytes": null, "top_logprobs": []} + ], + "refusal": null + })) + .unwrap(), + ); + chat_resp.response_events = vec![ + ResponseOutputEvent::Reasoning("think".into()), + ResponseOutputEvent::HiddenScores(1), + ResponseOutputEvent::Boundary(MessageBoundary::Start(MessagePhase::Commentary)), + ResponseOutputEvent::Text("Checking.".into()), + ResponseOutputEvent::Scores(1), + ResponseOutputEvent::Boundary(MessageBoundary::End), + ResponseOutputEvent::Boundary(MessageBoundary::Start(MessagePhase::FinalAnswer)), + ResponseOutputEvent::Text("Done.".into()), + ResponseOutputEvent::Scores(1), + ResponseOutputEvent::Boundary(MessageBoundary::End), + ResponseOutputEvent::Scores(1), + ]; + let params = ResponseParams { + include: Some(vec![IncludeEnum::MessageOutputTextLogprobs]), + top_logprobs: Some(0), + ..Default::default() + }; + + let resp = chat_completion_to_response(chat_resp, ¶ms, None).unwrap(); + let scores: Vec> = resp + .inner + .output + .iter() + .filter_map(|item| match item { + OutputItem::Message(message) => Some( + message + .content + .iter() + .filter_map(|part| match part { + OutputMessageContent::OutputText(text) => text.logprobs.as_ref(), + _ => None, + }) + .flatten() + .map(|entry| entry.token.clone()) + .collect(), + ), + _ => None, + }) + .collect(); + assert_eq!(scores, [vec!["Checking."], vec!["Done.", ""]]); + } + + /// Without message phases the score bookkeeping still drops reasoning scores + /// from the single text part. + #[test] + fn hidden_scores_are_dropped_without_phases() { + use crate::protocols::openai::chat_completions::ResponseOutputEvent; + + let mut chat_resp = make_chat_resp_with_text("hi"); + chat_resp.inner.choices[0].logprobs = Some( + serde_json::from_value(serde_json::json!({ + "content": [ + {"token": "think", "logprob": -0.1, "bytes": [116], "top_logprobs": []}, + {"token": "hi", "logprob": -0.25, "bytes": [104, 105], "top_logprobs": []}, + {"token": "", "logprob": -0.75, "bytes": null, "top_logprobs": []} + ], + "refusal": null + })) + .unwrap(), + ); + chat_resp.response_events = vec![ + ResponseOutputEvent::HiddenScores(1), + ResponseOutputEvent::Scores(2), + ]; + let params = ResponseParams { + include: Some(vec![IncludeEnum::MessageOutputTextLogprobs]), + ..Default::default() + }; + + let resp = chat_completion_to_response(chat_resp, ¶ms, None).unwrap(); + let OutputItem::Message(message) = &resp.inner.output[0] else { + panic!("expected message output"); + }; + let OutputMessageContent::OutputText(text) = &message.content[0] else { + panic!("expected text content"); + }; + let tokens: Vec<_> = text + .logprobs + .as_ref() + .unwrap() + .iter() + .map(|entry| entry.token.as_str()) + .collect(); + assert_eq!(tokens, ["hi", ""]); + } + #[test] fn chat_template_args_survive_the_conversion() { // The repro from the issue: a Responses request carrying template args diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index bf4285da85d8..b75cb3f4b91f 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -14,18 +14,19 @@ use std::time::{SystemTime, UNIX_EPOCH}; use axum::response::sse::Event; use dynamo_protocols::types::responses::{ - AssistantRole, IncompleteDetails, InputTokenDetails, Instructions, MessagePhase, OutputContent, - OutputItem, OutputMessage, OutputMessageContent, OutputStatus, OutputTextContent, - OutputTokenDetails, ReasoningItem, ReasoningItemContent, ReasoningTextContent, RefusalContent, - RequestToolChoiceParam, Response, ResponseCompletedEvent, ResponseContentPartAddedEvent, - ResponseContentPartDoneEvent, ResponseCreatedEvent, ResponseCustomToolCallInputDeltaEvent, - ResponseCustomToolCallInputDoneEvent, ResponseError, ResponseErrorCode, ResponseFailedEvent, - ResponseFunctionCallArgumentsDeltaEvent, ResponseFunctionCallArgumentsDoneEvent, - ResponseInProgressEvent, ResponseIncompleteEvent, ResponseOutputItemAddedEvent, - ResponseOutputItemDoneEvent, ResponseReasoningTextDeltaEvent, ResponseReasoningTextDoneEvent, - ResponseRefusalDeltaEvent, ResponseRefusalDoneEvent, ResponseStreamEvent, - ResponseTextDeltaEvent, ResponseTextDoneEvent, ResponseTextParam, ResponseUsage, - ServiceTierResponses, Status, TextResponseFormatConfiguration, ToolChoiceOptions, Truncation, + AssistantRole, IncompleteDetails, InputTokenDetails, Instructions, LogProb, MessagePhase, + OutputContent, OutputItem, OutputMessage, OutputMessageContent, OutputStatus, + OutputTextContent, OutputTokenDetails, ReasoningItem, ReasoningItemContent, + ReasoningTextContent, RefusalContent, RequestToolChoiceParam, Response, ResponseCompletedEvent, + ResponseContentPartAddedEvent, ResponseContentPartDoneEvent, ResponseCreatedEvent, + ResponseCustomToolCallInputDeltaEvent, ResponseCustomToolCallInputDoneEvent, ResponseError, + ResponseErrorCode, ResponseFailedEvent, ResponseFunctionCallArgumentsDeltaEvent, + ResponseFunctionCallArgumentsDoneEvent, ResponseInProgressEvent, ResponseIncompleteEvent, + ResponseOutputItemAddedEvent, ResponseOutputItemDoneEvent, ResponseReasoningTextDeltaEvent, + ResponseReasoningTextDoneEvent, ResponseRefusalDeltaEvent, ResponseRefusalDoneEvent, + ResponseStreamEvent, ResponseTextDeltaEvent, ResponseTextDoneEvent, ResponseTextParam, + ResponseUsage, ServiceTierResponses, Status, TextResponseFormatConfiguration, + ToolChoiceOptions, Truncation, }; use serde::{ Serialize, @@ -33,9 +34,12 @@ use serde::{ }; use uuid::Uuid; -use dynamo_protocols::types::{ChatCompletionMessageContent, FinishReason}; +use dynamo_protocols::types::{ChatChoiceStream, ChatCompletionMessageContent, FinishReason}; -use super::{ResponseParams, responses_incomplete_reason}; +use super::{ + ResponseParams, chat_logprobs_to_output_text, output_text_logprobs_to_events, + responses_incomplete_reason, +}; use crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse; use crate::protocols::unified::ResponsesContext; @@ -61,6 +65,19 @@ pub struct ResponseStreamConverter { accumulated_refusal: String, text_content_index: Option, refusal_content_index: Option, + // Scores received since the last text delta. Chunks that show no text + // (reasoning, tool calls, a trailing EOS score) still carry scores, which + // ride on the next text delta or on `output_text.done`. + pending_logprobs: Vec, + // Scores already delivered for the open message's text part. + text_logprobs: Vec, + // A `MessageBoundary::End` arrived while scores are requested. The message + // stays open so a later score-only chunk (hidden EOS) still joins it; the + // next boundary, visible content, or the end of the stream closes it. + close_deferred: bool, + // The message as published in `output_item.done`, reused for the closed + // list and the terminal response instead of rebuilding its scores. + finished_message: Option, // Ordered reasoning spans; a new item opens when reasoning resumes after // a tool call. reasoning_items: Vec, @@ -157,6 +174,10 @@ impl ResponseStreamConverter { accumulated_refusal: String::new(), text_content_index: None, refusal_content_index: None, + pending_logprobs: Vec::new(), + text_logprobs: Vec::new(), + close_deferred: false, + finished_message: None, reasoning_items: Vec::new(), active_reasoning_index: None, function_call_items: Vec::new(), @@ -209,6 +230,37 @@ impl ResponseStreamConverter { } } + /// Queue a choice's scores in arrival order; the next text delta, or the + /// close of the open message, carries them. Scores on reasoning or tool-call + /// chunks are not output text and are dropped; scores on chunks without + /// content (the hidden EOS) stay with the text they end. + fn queue_choice_logprobs(&mut self, choice: &ChatChoiceStream) { + let delta = &choice.delta; + let has_text = matches!( + delta.content.as_ref(), + Some(ChatCompletionMessageContent::Text(text)) if !text.is_empty() + ); + let hidden = !has_text + && (delta + .reasoning_content + .as_deref() + .is_some_and(|text| !text.is_empty()) + || delta + .tool_calls + .as_ref() + .is_some_and(|calls| !calls.is_empty())); + if self.params.logprobs_requested() + && !hidden + && let Some(entries) = choice + .logprobs + .as_ref() + .and_then(|lp| lp.content.as_deref()) + { + self.pending_logprobs + .extend(chat_logprobs_to_output_text(entries)); + } + } + fn append_message_part( &mut self, part: OutputContent, @@ -427,7 +479,7 @@ impl ResponseStreamConverter { .clone() .unwrap_or(ServiceTierResponses::Auto), ), - top_logprobs: Some(0), + top_logprobs: Some(self.params.echoed_top_logprobs()), usage: self.usage.clone(), } } @@ -543,6 +595,23 @@ impl ResponseStreamConverter { } } } + // A message whose close was deferred for trailing scores closes before + // anything that belongs to the next item arrives. + let chunk_has_content = chunk.inner.choices.iter().any(|choice| { + matches!(choice.delta.content.as_ref(), Some(ChatCompletionMessageContent::Text(text)) if !text.is_empty()) + || choice.delta.refusal.as_deref().is_some_and(|text| !text.is_empty()) + || choice.delta.reasoning_content.as_deref().is_some_and(|text| !text.is_empty()) + || choice.delta.tool_calls.as_ref().is_some_and(|calls| !calls.is_empty()) + }); + if self.close_deferred + && (chunk_has_content + || matches!( + &chunk.message_boundary, + Some(super::super::chat_completions::MessageBoundary::Start(_)) + )) + { + self.close_current_message(events); + } let mut should_finish_function_calls = false; match &chunk.message_boundary { Some(super::super::chat_completions::MessageBoundary::Start(phase)) => { @@ -558,13 +627,21 @@ impl ResponseStreamConverter { self.start_message(events); } Some(super::super::chat_completions::MessageBoundary::End) => { - self.close_current_message(events); + // Scores that still belong to this message (its EOS) can arrive + // after the boundary, so keep it open until the next item. + if self.params.logprobs_requested() && self.message_started { + self.close_deferred = true; + } else { + self.close_current_message(events); + } } None => {} } for choice in &chunk.inner.choices { let delta = &choice.delta; + self.queue_choice_logprobs(choice); + if let Some(reasoning) = delta.reasoning_content.as_deref() && !reasoning.is_empty() && !self.message_started @@ -608,6 +685,7 @@ impl ResponseStreamConverter { // Emit text delta self.accumulated_text.push_str(content); + let delta_logprobs = std::mem::take(&mut self.pending_logprobs); let text_delta = ResponseStreamEvent::ResponseOutputTextDelta(ResponseTextDeltaEvent { sequence_number: self.next_seq(), @@ -615,8 +693,9 @@ impl ResponseStreamConverter { output_index: self.message_output_index, content_index, delta: content.to_string(), - logprobs: Some(vec![]), + logprobs: Some(output_text_logprobs_to_events(&delta_logprobs)), }); + self.text_logprobs.extend(delta_logprobs); events.push(self.make_sse_event(&text_delta)); } @@ -1020,6 +1099,36 @@ impl ResponseStreamConverter { self.output_with_status(self.output_status()) } + /// The open message's parts for the close events; the token record moves + /// out instead of being copied. A message closed again (an error after its + /// completion) republishes the content it already published. + fn take_message_content(&mut self) -> Vec { + if let Some(finished) = &self.finished_message { + return finished.content.clone(); + } + let mut parts = Vec::with_capacity(2); + if let Some(index) = self.text_content_index { + parts.push(( + index, + OutputMessageContent::OutputText(OutputTextContent { + text: self.accumulated_text.clone(), + annotations: vec![], + logprobs: Some(std::mem::take(&mut self.text_logprobs)), + }), + )); + } + if let Some(index) = self.refusal_content_index { + parts.push(( + index, + OutputMessageContent::Refusal(RefusalContent { + refusal: self.accumulated_refusal.clone(), + }), + )); + } + parts.sort_unstable_by_key(|(index, _)| *index); + parts.into_iter().map(|(_, part)| part).collect() + } + fn message_content(&self) -> Vec { let mut parts = Vec::with_capacity(2); if let Some(index) = self.text_content_index { @@ -1028,7 +1137,7 @@ impl ResponseStreamConverter { OutputMessageContent::OutputText(OutputTextContent { text: self.accumulated_text.clone(), annotations: vec![], - logprobs: Some(vec![]), + logprobs: Some(self.text_logprobs.clone()), }), )); } @@ -1045,6 +1154,12 @@ impl ResponseStreamConverter { } fn message_output(&self, output_status: OutputStatus) -> OutputMessage { + // After the close events the published message is reused as-is. + if let Some(finished) = &self.finished_message { + let mut message = finished.clone(); + message.status = self.message_output_status.unwrap_or(output_status); + return message; + } OutputMessage { id: self.message_item_id.clone(), content: self.message_content(), @@ -1127,10 +1242,18 @@ impl ResponseStreamConverter { ); self.text_content_index = Some(index); } + // Scores that arrived after the last text delta (a tool call, a hidden + // EOS, a late score-only chunk) complete this text part's record. + let late_logprobs = std::mem::take(&mut self.pending_logprobs); + self.text_logprobs.extend(late_logprobs); - for (index, content) in self.message_content().into_iter().enumerate() { + // The parts are built once and travel through the events by value: each + // event hands its payload back after serialization, so the token record + // is not copied per event. + let mut content = Vec::with_capacity(2); + for (index, part) in self.take_message_content().into_iter().enumerate() { let content_index = index as u32; - let part = match content { + let part = match part { OutputMessageContent::OutputText(text) => { let done = ResponseStreamEvent::ResponseOutputTextDone(ResponseTextDoneEvent { sequence_number: self.next_seq(), @@ -1138,7 +1261,9 @@ impl ResponseStreamConverter { output_index: self.message_output_index, content_index, text: text.text.clone(), - logprobs: Some(vec![]), + logprobs: Some(output_text_logprobs_to_events( + text.logprobs.as_deref().unwrap_or_default(), + )), }); events.push(self.make_sse_event(&done)); OutputContent::OutputText(text) @@ -1163,14 +1288,39 @@ impl ResponseStreamConverter { part, }); events.push(self.make_sse_event(&done)); + let ResponseStreamEvent::ResponseContentPartDone(done) = done else { + unreachable!("content part event was just built") + }; + content.push(match done.part { + OutputContent::OutputText(text) => OutputMessageContent::OutputText(text), + OutputContent::Refusal(refusal) => OutputMessageContent::Refusal(refusal), + OutputContent::ReasoningText(_) => { + unreachable!("message parts are text or refusal") + } + }); } + let message = OutputMessage { + id: self.message_item_id.clone(), + content, + role: AssistantRole::Assistant, + phase: self.message_phase, + status: self.message_output_status.unwrap_or(output_status), + }; let item_done = ResponseStreamEvent::ResponseOutputItemDone(ResponseOutputItemDoneEvent { sequence_number: self.next_seq(), output_index: self.message_output_index, - item: OutputItem::Message(self.message_output(output_status)), + item: OutputItem::Message(message), }); events.push(self.make_sse_event(&item_done)); + let ResponseStreamEvent::ResponseOutputItemDone(ResponseOutputItemDoneEvent { + item: OutputItem::Message(message), + .. + }) = item_done + else { + unreachable!("output item event was just built") + }; + self.finished_message = Some(message); } fn close_current_message(&mut self, events: &mut Vec>) { @@ -1178,10 +1328,12 @@ impl ResponseStreamConverter { return; } self.close_open_message_item(events, OutputStatus::Completed); - self.closed_messages.push(( - self.message_output_index, - self.message_output(OutputStatus::Completed), - )); + let message = self + .finished_message + .take() + .unwrap_or_else(|| self.message_output(OutputStatus::Completed)); + self.closed_messages + .push((self.message_output_index, message)); self.message_item_id = format!("msg_{}", Uuid::new_v4().simple()); self.message_started = false; self.message_output_status = None; @@ -1190,6 +1342,8 @@ impl ResponseStreamConverter { self.accumulated_refusal.clear(); self.text_content_index = None; self.refusal_content_index = None; + self.text_logprobs.clear(); + self.close_deferred = false; } /// Emit remaining output completion events and `response.completed` at stream end. @@ -1210,6 +1364,11 @@ impl ResponseStreamConverter { // whether the still-open reasoning item completed or was truncated. self.append_active_reasoning_done_events(events, output_status); + // A message that ended at a parser boundary completed; it only waited for + // trailing scores, which the close now drains into it. + if self.close_deferred { + self.close_current_message(events); + } // Only the terminal item was cut short, and the terminal response reports it // that way. The `output_item.done` event has to agree, or a client sees one // item marked incomplete in the stream and completed in the final response. @@ -1264,6 +1423,9 @@ impl ResponseStreamConverter { ) -> Result { let output_status = OutputStatus::Incomplete; self.append_active_reasoning_done_events(events, output_status); + if self.close_deferred { + self.close_current_message(events); + } self.close_open_message_item(events, output_status); self.append_pending_function_call_done_events(events, output_status, false); @@ -1841,8 +2003,270 @@ mod tests { ); } + /// Decode the JSON body of one SSE event. `Event` exposes only its `Debug` + /// form, which prints the wire buffer as a Rust byte-string literal. + fn event_json(event: &Result) -> serde_json::Value { + let debug = format!("{:?}", event.as_ref().unwrap()); + let start = debug.find("data: ").expect("event has a data line") + 6; + let literal = &debug[start..]; + let mut json = String::new(); + let mut chars = literal.chars(); + while let Some(c) = chars.next() { + match c { + '\\' => match chars.next() { + Some('n') => break, + Some('"') => json.push('"'), + Some('\\') => json.push('\\'), + Some('t') => json.push('\t'), + Some('r') => json.push('\r'), + Some('\'') => json.push('\''), + Some('x') => { + let hex: String = chars.by_ref().take(2).collect(); + json.push(u8::from_str_radix(&hex, 16).unwrap() as char); + } + other => panic!("unexpected escape {other:?}"), + }, + _ => json.push(c), + } + } + serde_json::from_str(&json).expect("event data is JSON") + } + + fn logprob_tokens(logprobs: &serde_json::Value) -> Vec<&str> { + logprobs + .as_array() + .expect("logprobs is an array") + .iter() + .map(|entry| entry["token"].as_str().unwrap()) + .collect() + } + + fn scored( + mut chunk: NvCreateChatCompletionStreamResponse, + token: &str, + ) -> NvCreateChatCompletionStreamResponse { + chunk.inner.choices[0].logprobs = Some( + serde_json::from_value(serde_json::json!({ + "content": [{ + "token": token, + "logprob": -0.5, + "bytes": token.as_bytes(), + "top_logprobs": [{"token": token, "logprob": -0.5, "bytes": token.as_bytes()}] + }], + "refusal": null + })) + .unwrap(), + ); + chunk + } + + fn logprob_params() -> ResponseParams { + use dynamo_protocols::types::responses::IncludeEnum; + + ResponseParams { + include: Some(vec![IncludeEnum::MessageOutputTextLogprobs]), + top_logprobs: Some(1), + ..default_params() + } + } + + /// Every chat score reaches the client once, in generation order: a text + /// delta carries the scores received since the previous delta, and scores + /// after the last delta complete the record on `output_text.done`. + #[test] + fn requested_logprobs_ride_on_text_deltas_and_done() { + let mut score_only = finish_chunk(FinishReason::Stop); + score_only.inner.choices[0] = + crate::protocols::openai::chat_completions::unified_parser::empty_choice(0); + let score_only = scored(score_only, "D"); + + let mut conv = ResponseStreamConverter::new("test-model".into(), logprob_params()); + let _ = conv.emit_start_events(); + // Reasoning is not output text: neither its text nor its score is published. + assert!( + conv.process_chunk(&scored(reasoning_chunk("think"), "A")) + .is_empty() + ); + + let events = conv.process_chunk(&scored(text_chunk("hi"), "B")); + let delta = events + .iter() + .find(|event| event_type(event) == "response.output_text.delta") + .expect("text delta"); + assert_eq!(logprob_tokens(&event_json(delta)["logprobs"]), ["B"]); + + let events = conv.process_chunk(&scored(text_chunk("!"), "C")); + assert_eq!(logprob_tokens(&event_json(&events[0])["logprobs"]), ["C"]); + + conv.process_chunk(&finish_chunk(FinishReason::Stop)); + assert!(conv.process_chunk(&score_only).is_empty()); + + let events = conv.emit_end_events(); + let done = event_json(&events[0]); + assert_eq!(done["type"], "response.output_text.done"); + assert_eq!(logprob_tokens(&done["logprobs"]), ["B", "C", "D"]); + assert_eq!(done["logprobs"][0]["top_logprobs"][0]["token"], "B"); + + let completed = event_json(events.last().unwrap()); + assert_eq!(completed["response"]["top_logprobs"], 1); + let text = &completed["response"]["output"][0]["content"][0]; + assert_eq!(text["text"], "hi!"); + assert_eq!(logprob_tokens(&text["logprobs"]), ["B", "C", "D"]); + assert_eq!(text["logprobs"][0]["bytes"], serde_json::json!([66])); + } + + /// Scores on the chunk that ends a message belong to that message, and scores + /// on the chunk that starts the next one belong to the next. + #[test] + fn boundary_chunk_scores_stay_with_their_message() { + use super::super::super::chat_completions::MessageBoundary; + use dynamo_protocols::types::responses::MessagePhase; + + let mut end = finish_chunk(FinishReason::Stop); + end.inner.choices[0].finish_reason = None; + end.message_boundary = Some(MessageBoundary::End); + let end = scored(end, "B"); + let mut start = finish_chunk(FinishReason::Stop); + start.inner.choices[0].finish_reason = None; + start.message_boundary = Some(MessageBoundary::Start(MessagePhase::FinalAnswer)); + let start = scored(start, "C"); + + let mut conv = ResponseStreamConverter::new("test-model".into(), logprob_params()); + let _ = conv.emit_start_events(); + conv.process_chunk(&scored(text_chunk("first"), "A")); + conv.process_chunk(&end); + conv.process_chunk(&start); + conv.process_chunk(&scored(text_chunk("second"), "D")); + conv.process_chunk(&finish_chunk(FinishReason::Stop)); + let _ = conv.emit_end_events(); + + let messages: Vec> = conv + .completed_output() + .iter() + .filter_map(|item| match item { + OutputItem::Message(message) => Some( + message + .content + .iter() + .filter_map(|part| match part { + OutputMessageContent::OutputText(text) => text.logprobs.as_ref(), + _ => None, + }) + .flatten() + .map(|entry| entry.token.clone()) + .collect(), + ), + _ => None, + }) + .collect(); + assert_eq!(messages, [["A", "B"], ["C", "D"]]); + } + + /// A message closed by a parser `End` boundary still owns the EOS score that a + /// later score-only terminal chunk carries: its close waits for that chunk, so + /// `output_text.done` and the completed output both hold the full record. + #[test] + fn end_boundary_waits_for_trailing_scores() { + use super::super::super::chat_completions::MessageBoundary; + + let mut end = finish_chunk(FinishReason::Stop); + end.inner.choices[0].finish_reason = None; + end.message_boundary = Some(MessageBoundary::End); + let eos = scored(finish_chunk(FinishReason::Stop), "E"); + + let mut conv = ResponseStreamConverter::new("test-model".into(), logprob_params()); + let _ = conv.emit_start_events(); + conv.process_chunk(&scored(text_chunk("Done."), "A")); + // The boundary alone publishes nothing yet. + assert!(conv.process_chunk(&end).is_empty()); + assert!(conv.process_chunk(&eos).is_empty()); + + let events = conv.emit_end_events(); + assert_eq!( + event_types(&events), + [ + "response.output_text.done", + "response.content_part.done", + "response.output_item.done", + "response.completed", + ] + ); + let done = event_json(&events[0]); + assert_eq!(logprob_tokens(&done["logprobs"]), ["A", "E"]); + let item = event_json(&events[2]); + assert_eq!(item["item"]["status"], "completed"); + assert_eq!( + logprob_tokens(&item["item"]["content"][0]["logprobs"]), + ["A", "E"] + ); + let completed = event_json(&events[3]); + assert_eq!(completed["response"]["output"].as_array().unwrap().len(), 1); + assert_eq!( + logprob_tokens(&completed["response"]["output"][0]["content"][0]["logprobs"]), + ["A", "E"] + ); + } + + /// An error after the message completed closes it again; the republished + /// message keeps the scores it already published instead of an empty record. + #[test] + fn error_after_completion_keeps_published_scores() { + let mut conv = ResponseStreamConverter::new("test-model".into(), logprob_params()); + let _ = conv.emit_start_events(); + conv.process_chunk(&scored(text_chunk("Checking."), "A")); + conv.process_chunk(&finish_chunk(FinishReason::Stop)); + let events = conv.emit_end_events(); + let completed = event_json(events.last().unwrap()); + assert_eq!( + logprob_tokens(&completed["response"]["output"][0]["content"][0]["logprobs"]), + ["A"] + ); + + let events = conv.emit_error_events(ResponseError { + misalignment: None, + code: ResponseErrorCode::ServerError, + message: "tool arguments were malformed".into(), + }); + let done = event_json(&events[0]); + assert_eq!(done["type"], "response.output_text.done"); + assert_eq!(logprob_tokens(&done["logprobs"]), ["A"]); + let failed = event_json(events.last().unwrap()); + assert_eq!(failed["type"], "response.failed"); + assert_eq!( + logprob_tokens(&failed["response"]["output"][0]["content"][0]["logprobs"]), + ["A"] + ); + } + + /// Without `include: message.output_text.logprobs`, chat scores are dropped and + /// the schema-required arrays stay empty. + #[test] + fn logprobs_stay_empty_without_include() { + let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); + let _ = conv.emit_start_events(); + + let events = conv.process_chunk(&scored(text_chunk("hi"), "A")); + let delta = events + .iter() + .find(|event| event_type(event) == "response.output_text.delta") + .expect("text delta"); + assert_eq!(event_json(delta)["logprobs"], serde_json::json!([])); + + conv.process_chunk(&finish_chunk(FinishReason::Stop)); + let events = conv.emit_end_events(); + let done = event_json(&events[0]); + assert_eq!(done["logprobs"], serde_json::json!([])); + let completed = event_json(events.last().unwrap()); + assert_eq!(completed["response"]["top_logprobs"], 0); + assert_eq!( + completed["response"]["output"][0]["content"][0]["logprobs"], + serde_json::json!([]) + ); + } + /// The V1 tool jail can send late scores on an empty-delta choice after `finish_reason`. - /// Responses carries no chat logprobs, so that choice must not change the output. + /// Without an `include` request Responses drops chat logprobs, so that choice must not + /// change the output. #[test] fn post_finish_score_only_choice_emits_no_events() { let mut score_only = finish_chunk(FinishReason::Stop); diff --git a/lib/llm/tests/data/replays/agent-harness/text-logprobs.sse b/lib/llm/tests/data/replays/agent-harness/text-logprobs.sse new file mode 100644 index 000000000000..8fdd81ff75da --- /dev/null +++ b/lib/llm/tests/data/replays/agent-harness/text-logprobs.sse @@ -0,0 +1,11 @@ +data: {"id":"chatcmpl-harness-text-logprobs","object":"chat.completion.chunk","created":1000000000,"model":"harness-model","choices":[{"index":0,"delta":{"role":"assistant","content":""}}]} + +data: {"id":"chatcmpl-harness-text-logprobs","object":"chat.completion.chunk","created":1000000000,"model":"harness-model","choices":[{"index":0,"delta":{"content":"Pong"},"logprobs":{"content":[{"token":"Pong","logprob":-0.25,"bytes":[80,111,110,103],"top_logprobs":[{"token":"Pong","logprob":-0.25,"bytes":[80,111,110,103]},{"token":"Ping","logprob":-1.5,"bytes":[80,105,110,103]}]}],"refusal":null}}]} + +data: {"id":"chatcmpl-harness-text-logprobs","object":"chat.completion.chunk","created":1000000000,"model":"harness-model","choices":[{"index":0,"delta":{"content":"."},"logprobs":{"content":[{"token":".","logprob":-0.5,"bytes":[46],"top_logprobs":[{"token":".","logprob":-0.5,"bytes":[46]},{"token":"!","logprob":-2.0,"bytes":[33]}]}],"refusal":null}}]} + +data: {"id":"chatcmpl-harness-text-logprobs","object":"chat.completion.chunk","created":1000000000,"model":"harness-model","choices":[{"index":0,"delta":{},"finish_reason":"stop","logprobs":{"content":[{"token":"","logprob":-0.75,"bytes":null,"top_logprobs":[]}],"refusal":null}}]} + +data: {"id":"chatcmpl-harness-text-logprobs","object":"chat.completion.chunk","created":1000000000,"model":"harness-model","choices":[],"usage":{"prompt_tokens":5,"completion_tokens":3,"total_tokens":8}} + +data: [DONE] diff --git a/lib/llm/tests/responses_http_replay.rs b/lib/llm/tests/responses_http_replay.rs index a0787983de95..ce6d01dab5c3 100644 --- a/lib/llm/tests/responses_http_replay.rs +++ b/lib/llm/tests/responses_http_replay.rs @@ -620,6 +620,140 @@ async fn streaming_text_baseline() { .await; } +fn logprob_tokens(logprobs: &Value) -> Vec<&str> { + logprobs + .as_array() + .expect("logprobs is an array") + .iter() + .map(|entry| entry["token"].as_str().unwrap()) + .collect() +} + +/// `include: message.output_text.logprobs` turns on backend scoring and puts every +/// chat token record, including the hidden EOS score on the finish chunk, on the +/// output text part; the response echoes the requested `top_logprobs`. +#[tokio::test] +#[serial] +async fn unary_text_logprobs_reach_output_text() { + temp_env::async_with_vars(ENV, async { + let svc = HarnessService::start([ + load_agent_fixture("text-logprobs.sse").await.unwrap(), + load_agent_fixture("text-logprobs.sse").await.unwrap(), + ]) + .await; + + let response = post_responses( + &svc, + &json!({ + "model": MODEL, + "input": "ping", + "stream": false, + "max_output_tokens": 64, + "include": ["message.output_text.logprobs"], + "top_logprobs": 2 + }), + ) + .await; + assert_eq!(response.status(), reqwest::StatusCode::OK); + let body: Value = response.json().await.unwrap(); + let text = &body["output"][0]["content"][0]; + assert_eq!(text["text"], "Pong."); + assert_eq!(logprob_tokens(&text["logprobs"]), ["Pong", ".", ""]); + assert_eq!(text["logprobs"][0]["logprob"], -0.25); + assert_eq!(text["logprobs"][0]["bytes"], json!([80, 111, 110, 103])); + assert_eq!(text["logprobs"][0]["top_logprobs"][1]["token"], "Ping"); + assert_eq!(text["logprobs"][2]["bytes"], json!([])); + assert_eq!(body["top_logprobs"], 2); + + // Without the include selector the arrays stay empty and nothing is + // requested from the backend. + let response = post_responses( + &svc, + &json!({ + "model": MODEL, + "input": "ping", + "stream": false, + "max_output_tokens": 64, + "top_logprobs": 2 + }), + ) + .await; + assert_eq!(response.status(), reqwest::StatusCode::OK); + let body: Value = response.json().await.unwrap(); + assert_eq!(body["output"][0]["content"][0]["logprobs"], json!([])); + assert_eq!(body["top_logprobs"], 2); + + let requests = svc.engine.take_requests().await; + assert_eq!(requests.len(), 2); + assert_eq!(requests[0].inner.logprobs, Some(true)); + assert_eq!(requests[0].inner.top_logprobs, Some(2)); + assert_eq!(requests[1].inner.logprobs, None); + assert_eq!(requests[1].inner.top_logprobs, None); + assert_eq!(svc.engine.remaining_scripts().await, 0); + svc.shutdown().await; + }) + .await; +} + +/// Each text delta carries the scores received since the previous delta; the +/// EOS score on the finish chunk completes the record on `output_text.done` +/// and on the completed output. +#[tokio::test] +#[serial] +async fn streaming_text_logprobs_ride_on_deltas_and_done() { + temp_env::async_with_vars(ENV, async { + let svc = + HarnessService::start([load_agent_fixture("text-logprobs.sse").await.unwrap()]).await; + let response = post_responses( + &svc, + &json!({ + "model": MODEL, + "input": "ping", + "stream": true, + "max_output_tokens": 64, + "include": ["message.output_text.logprobs"], + "top_logprobs": 2 + }), + ) + .await; + assert_eq!(response.status(), reqwest::StatusCode::OK); + let raw = response.text().await.unwrap(); + let events = parse_responses_sse(&raw).await.unwrap(); + + let deltas: Vec<_> = events + .iter() + .filter(|event| event.event == "response.output_text.delta") + .collect(); + assert_eq!(deltas.len(), 2); + assert_eq!(logprob_tokens(&deltas[0].data["logprobs"]), ["Pong"]); + assert_eq!( + deltas[0].data["logprobs"][0]["top_logprobs"][1]["token"], + "Ping" + ); + assert_eq!(logprob_tokens(&deltas[1].data["logprobs"]), ["."]); + + let done = events + .iter() + .find(|event| event.event == "response.output_text.done") + .expect("output_text.done"); + assert_eq!(done.data["text"], "Pong."); + assert_eq!(logprob_tokens(&done.data["logprobs"]), ["Pong", ".", ""]); + + let completed = events + .iter() + .find(|event| event.event == "response.completed") + .expect("response.completed"); + let text = &completed.data["response"]["output"][0]["content"][0]; + assert_eq!(logprob_tokens(&text["logprobs"]), ["Pong", ".", ""]); + assert_eq!(text["logprobs"][2]["bytes"], json!([])); + assert_eq!(completed.data["response"]["top_logprobs"], 2); + + assert_eq!(svc.engine.remaining_scripts().await, 0); + svc.shutdown().await; + }) + .await; +} + #[rstest::rstest] #[case::text(false)] #[case::refusal(true)] diff --git a/tests/serve/test_vllm.py b/tests/serve/test_vllm.py index 75394243f17e..e1afd8f85e3c 100644 --- a/tests/serve/test_vllm.py +++ b/tests/serve/test_vllm.py @@ -37,6 +37,8 @@ lora_chat_payload, metric_payload_default, pooling_payload, + responses_payload_with_logprobs, + responses_stream_payload_with_logprobs, router_cached_tokens_chat_payload, router_selection_chat_payload_default, ) @@ -223,6 +225,23 @@ class VLLMConfig(EngineConfig): temperature=0.0, logprobs=5, ), + # Responses API scores ride on output_text. Thinking is disabled so the + # budget produces visible text, and the budget lets the answer finish, + # since the Responses payload requires status "completed". + responses_payload_with_logprobs( + expected_response=["AI", "knock", "joke"], + max_tokens=200, + temperature=0.0, + top_logprobs=3, + extra_body={"chat_template_args": {"enable_thinking": False}}, + ), + responses_stream_payload_with_logprobs( + expected_response=["AI", "knock", "joke"], + max_tokens=200, + temperature=0.0, + top_logprobs=3, + extra_body={"chat_template_args": {"enable_thinking": False}}, + ), ], ), "aggregated_lmcache": VLLMConfig( diff --git a/tests/utils/payload_builder.py b/tests/utils/payload_builder.py index 8fe3fbb301fe..7bc20aa9e624 100644 --- a/tests/utils/payload_builder.py +++ b/tests/utils/payload_builder.py @@ -34,7 +34,9 @@ MetricsPayload, PoolingPayload, ResponsesPayload, + ResponsesPayloadWithLogprobs, ResponsesStreamPayload, + ResponsesStreamPayloadWithLogprobs, RouterNvextChatPayload, SGLangGenerateCancellationPayload, SGLangGenerateRecoveryPayload, @@ -1253,6 +1255,59 @@ def responses_stream_payload_default( ) +def responses_payload_with_logprobs( + repeat_count: int = 1, + expected_response: Optional[List[str]] = None, + max_tokens: int = 50, + temperature: float = 0.0, + top_logprobs: int = 3, + extra_body: Optional[Dict[str, Any]] = None, +) -> ResponsesPayloadWithLogprobs: + """Create a non-streaming Responses payload that requests output logprobs.""" + body: Dict[str, Any] = { + "input": TEXT_PROMPT, + "max_output_tokens": max_tokens, + "temperature": temperature, + "include": ["message.output_text.logprobs"], + "top_logprobs": top_logprobs, + } + if extra_body: + body.update(extra_body) + return ResponsesPayloadWithLogprobs( + body=body, + repeat_count=repeat_count, + expected_log=[], + expected_response=expected_response or ["AI", "knock", "joke"], + ) + + +def responses_stream_payload_with_logprobs( + repeat_count: int = 1, + expected_response: Optional[List[str]] = None, + max_tokens: int = 50, + temperature: float = 0.0, + top_logprobs: int = 3, + extra_body: Optional[Dict[str, Any]] = None, +) -> ResponsesStreamPayloadWithLogprobs: + """Create a streaming Responses payload that requests output logprobs.""" + body: Dict[str, Any] = { + "input": TEXT_PROMPT, + "stream": True, + "max_output_tokens": max_tokens, + "temperature": temperature, + "include": ["message.output_text.logprobs"], + "top_logprobs": top_logprobs, + } + if extra_body: + body.update(extra_body) + return ResponsesStreamPayloadWithLogprobs( + body=body, + repeat_count=repeat_count, + expected_log=[], + expected_response=expected_response or ["AI", "knock", "joke"], + ) + + def anthropic_messages_payload_default( repeat_count: int = 1, expected_response: Optional[List[str]] = None, diff --git a/tests/utils/payloads.py b/tests/utils/payloads.py index 4ee9769d3ee8..b45a57cc8880 100644 --- a/tests/utils/payloads.py +++ b/tests/utils/payloads.py @@ -2166,6 +2166,116 @@ def response_handler(self, response: Any) -> str: return ResponsesStreamPayload.extract_content(response) +def _validate_responses_logprobs(entries: Any, top_logprobs: int) -> None: + """Check `output_text.logprobs` entries from the Responses API.""" + assert entries, "Expected output_text logprobs when include requests them" + for entry in entries: + assert isinstance(entry["token"], str), f"Invalid token: {entry!r}" + value = entry["logprob"] + assert math.isfinite(value) and value <= 0, f"Invalid logprob: {entry!r}" + # Output parts carry `bytes`; the text stream events do not. + if "bytes" in entry: + assert isinstance(entry["bytes"], list), f"Invalid token bytes: {entry!r}" + assert ( + len(entry["top_logprobs"]) <= top_logprobs + ), f"More alternatives than requested: {entry!r}" + for top in entry["top_logprobs"]: + assert isinstance(top["token"], str), f"Invalid top token: {top!r}" + assert math.isfinite(top["logprob"]) and top["logprob"] <= 0 + + +@dataclass +class ResponsesPayloadWithLogprobs(ResponsesPayload): + """Non-streaming Responses payload that requests and validates output logprobs. + + One score per generated token must reach `output_text.logprobs`, so the entry + count is checked against `usage.output_tokens`. + """ + + def validate(self, response: Any, content: str) -> None: + super().validate(response, content) + result = response.json() + top_logprobs = self.body.get("top_logprobs") or 0 + assert ( + result.get("top_logprobs") == top_logprobs + ), f"top_logprobs echo mismatch: {result.get('top_logprobs')} != {top_logprobs}" + text_part = result["output"][0]["content"][0] + entries = text_part.get("logprobs") + _validate_responses_logprobs(entries, top_logprobs) + output_tokens = result["usage"]["output_tokens"] + assert ( + len(entries) == output_tokens + ), f"Expected {output_tokens} logprob entries, got {len(entries)}" + logger.info( + "Responses logprobs validation passed: %d entries, top_logprobs=%d", + len(entries), + top_logprobs, + ) + + +@dataclass +class ResponsesStreamPayloadWithLogprobs(ResponsesStreamPayload): + """Streaming Responses payload that validates logprobs on text events. + + Every `output_text.delta` carries a logprobs array; `output_text.done` and the + completed output carry the full record, which starts with the concatenated delta + scores and may end with scores that arrived after the last delta (hidden EOS). + """ + + def response_handler(self, response: Any) -> str: + response.raise_for_status() + events: list[tuple[str, Any]] = [] + event_type = "" + for line in response.iter_lines(decode_unicode=True): + if not line: + continue + if line.startswith("event: "): + event_type = line[len("event: ") :] + elif line.startswith("data: "): + data_str = line[len("data: ") :] + if data_str != "[DONE]": + events.append((event_type, json.loads(data_str))) + + top_logprobs = self.body.get("top_logprobs") or 0 + deltas = [e for t, e in events if t == "response.output_text.delta"] + assert deltas, "Missing output_text.delta events" + assert all( + isinstance(d.get("logprobs"), list) for d in deltas + ), "Every output_text.delta must carry a logprobs array" + delta_tokens = [lp["token"] for d in deltas for lp in d["logprobs"]] + assert delta_tokens, "No logprobs on any output_text.delta" + + done_events = [e for t, e in events if t == "response.output_text.done"] + assert ( + len(done_events) == 1 + ), f"Expected 1 output_text.done, got {len(done_events)}" + done_entries = done_events[0]["logprobs"] + _validate_responses_logprobs(done_entries, top_logprobs) + done_tokens = [lp["token"] for lp in done_entries] + assert ( + done_tokens[: len(delta_tokens)] == delta_tokens + ), "output_text.done logprobs must start with the delta logprobs" + + completed = [e for t, e in events if t == "response.completed"] + assert ( + len(completed) == 1 + ), f"Expected 1 response.completed, got {len(completed)}" + final = completed[0]["response"] + assert final.get("top_logprobs") == top_logprobs, "top_logprobs echo mismatch" + final_entries = final["output"][0]["content"][0]["logprobs"] + assert [lp["token"] for lp in final_entries] == done_tokens + output_tokens = final["usage"]["output_tokens"] + assert ( + len(done_tokens) == output_tokens + ), f"Expected {output_tokens} logprob entries, got {len(done_tokens)}" + logger.info( + "Responses stream logprobs validation passed: %d on deltas, %d total", + len(delta_tokens), + len(done_tokens), + ) + return "".join(d["delta"] for d in deltas) + + @dataclass class AnthropicMessagesPayload(BasePayload): """Payload for the Anthropic Messages API endpoint (/v1/messages)."""