revert /reply to previous behavior (replacing session history) when full conversation provided (#6058)
This commit is contained in:
@@ -75,7 +75,9 @@ fn track_tool_telemetry(content: &MessageContent, all_messages: &[Message]) {
|
|||||||
|
|
||||||
#[derive(Debug, Deserialize, Serialize, utoipa::ToSchema)]
|
#[derive(Debug, Deserialize, Serialize, utoipa::ToSchema)]
|
||||||
pub struct ChatRequest {
|
pub struct ChatRequest {
|
||||||
messages: Vec<Message>,
|
user_message: Message,
|
||||||
|
#[serde(default)]
|
||||||
|
conversation_so_far: Option<Vec<Message>>,
|
||||||
session_id: String,
|
session_id: String,
|
||||||
recipe_name: Option<String>,
|
recipe_name: Option<String>,
|
||||||
recipe_version: Option<String>,
|
recipe_version: Option<String>,
|
||||||
@@ -233,7 +235,8 @@ pub async fn reply(
|
|||||||
let stream = ReceiverStream::new(rx);
|
let stream = ReceiverStream::new(rx);
|
||||||
let cancel_token = CancellationToken::new();
|
let cancel_token = CancellationToken::new();
|
||||||
|
|
||||||
let messages = Conversation::new_unvalidated(request.messages);
|
let user_message = request.user_message;
|
||||||
|
let conversation_so_far = request.conversation_so_far;
|
||||||
|
|
||||||
let task_cancel = cancel_token.clone();
|
let task_cancel = cancel_token.clone();
|
||||||
let task_tx = tx.clone();
|
let task_tx = tx.clone();
|
||||||
@@ -255,7 +258,7 @@ pub async fn reply(
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
let session = match SessionManager::get_session(&session_id, false).await {
|
let session = match SessionManager::get_session(&session_id, true).await {
|
||||||
Ok(metadata) => metadata,
|
Ok(metadata) => metadata,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::error!("Failed to read session for {}: {}", session_id, e);
|
tracing::error!("Failed to read session for {}: {}", session_id, e);
|
||||||
@@ -278,20 +281,21 @@ pub async fn reply(
|
|||||||
retry_config: None,
|
retry_config: None,
|
||||||
};
|
};
|
||||||
|
|
||||||
let user_message = match messages.last() {
|
let mut all_messages = match conversation_so_far {
|
||||||
Some(msg) => msg,
|
Some(history) => {
|
||||||
_ => {
|
let conv = Conversation::new_unvalidated(history);
|
||||||
let _ = stream_event(
|
if let Err(e) = SessionManager::replace_conversation(&session_id, &conv).await {
|
||||||
MessageEvent::Error {
|
tracing::warn!(
|
||||||
error: "Reply started with empty messages".to_string(),
|
"Failed to replace session conversation for {}: {}",
|
||||||
},
|
session_id,
|
||||||
&task_tx,
|
e
|
||||||
&task_cancel,
|
);
|
||||||
)
|
}
|
||||||
.await;
|
conv
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
|
None => session.conversation.unwrap_or_default(),
|
||||||
};
|
};
|
||||||
|
all_messages.push(user_message.clone());
|
||||||
|
|
||||||
let mut stream = match agent
|
let mut stream = match agent
|
||||||
.reply(
|
.reply(
|
||||||
@@ -316,8 +320,6 @@ pub async fn reply(
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut all_messages = messages.clone();
|
|
||||||
|
|
||||||
let mut heartbeat_interval = tokio::time::interval(Duration::from_millis(500));
|
let mut heartbeat_interval = tokio::time::interval(Duration::from_millis(500));
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
@@ -478,7 +480,8 @@ mod tests {
|
|||||||
.header("x-secret-key", "test-secret")
|
.header("x-secret-key", "test-secret")
|
||||||
.body(Body::from(
|
.body(Body::from(
|
||||||
serde_json::to_string(&ChatRequest {
|
serde_json::to_string(&ChatRequest {
|
||||||
messages: vec![Message::user().with_text("test message")],
|
user_message: Message::user().with_text("test message"),
|
||||||
|
conversation_so_far: None,
|
||||||
session_id: "test-session".to_string(),
|
session_id: "test-session".to_string(),
|
||||||
recipe_name: None,
|
recipe_name: None,
|
||||||
recipe_version: None,
|
recipe_version: None,
|
||||||
|
|||||||
@@ -2715,15 +2715,16 @@
|
|||||||
"ChatRequest": {
|
"ChatRequest": {
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"required": [
|
"required": [
|
||||||
"messages",
|
"user_message",
|
||||||
"session_id"
|
"session_id"
|
||||||
],
|
],
|
||||||
"properties": {
|
"properties": {
|
||||||
"messages": {
|
"conversation_so_far": {
|
||||||
"type": "array",
|
"type": "array",
|
||||||
"items": {
|
"items": {
|
||||||
"$ref": "#/components/schemas/Message"
|
"$ref": "#/components/schemas/Message"
|
||||||
}
|
},
|
||||||
|
"nullable": true
|
||||||
},
|
},
|
||||||
"recipe_name": {
|
"recipe_name": {
|
||||||
"type": "string",
|
"type": "string",
|
||||||
@@ -2735,6 +2736,9 @@
|
|||||||
},
|
},
|
||||||
"session_id": {
|
"session_id": {
|
||||||
"type": "string"
|
"type": "string"
|
||||||
|
},
|
||||||
|
"user_message": {
|
||||||
|
"$ref": "#/components/schemas/Message"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -59,10 +59,11 @@ export type CallToolResponse = {
|
|||||||
};
|
};
|
||||||
|
|
||||||
export type ChatRequest = {
|
export type ChatRequest = {
|
||||||
messages: Array<Message>;
|
conversation_so_far?: Array<Message> | null;
|
||||||
recipe_name?: string | null;
|
recipe_name?: string | null;
|
||||||
recipe_version?: string | null;
|
recipe_version?: string | null;
|
||||||
session_id: string;
|
session_id: string;
|
||||||
|
user_message: Message;
|
||||||
};
|
};
|
||||||
|
|
||||||
export type CheckProviderRequest = {
|
export type CheckProviderRequest = {
|
||||||
|
|||||||
@@ -292,12 +292,13 @@ export function useChatStream({
|
|||||||
window.dispatchEvent(new CustomEvent('session-created'));
|
window.dispatchEvent(new CustomEvent('session-created'));
|
||||||
}
|
}
|
||||||
|
|
||||||
// Build message list: add new message if provided, otherwise continue with existing
|
const newMessage = hasNewMessage
|
||||||
|
? createUserMessage(userMessage)
|
||||||
|
: messagesRef.current[messagesRef.current.length - 1];
|
||||||
const currentMessages = hasNewMessage
|
const currentMessages = hasNewMessage
|
||||||
? [...messagesRef.current, createUserMessage(userMessage)]
|
? [...messagesRef.current, newMessage]
|
||||||
: [...messagesRef.current];
|
: [...messagesRef.current];
|
||||||
|
|
||||||
// Update UI with new message before streaming
|
|
||||||
if (hasNewMessage) {
|
if (hasNewMessage) {
|
||||||
updateMessages(currentMessages);
|
updateMessages(currentMessages);
|
||||||
}
|
}
|
||||||
@@ -310,7 +311,7 @@ export function useChatStream({
|
|||||||
const { stream } = await reply({
|
const { stream } = await reply({
|
||||||
body: {
|
body: {
|
||||||
session_id: sessionId,
|
session_id: sessionId,
|
||||||
messages: currentMessages,
|
user_message: newMessage,
|
||||||
},
|
},
|
||||||
throwOnError: true,
|
throwOnError: true,
|
||||||
signal: abortControllerRef.current.signal,
|
signal: abortControllerRef.current.signal,
|
||||||
@@ -356,7 +357,7 @@ export function useChatStream({
|
|||||||
const { stream } = await reply({
|
const { stream } = await reply({
|
||||||
body: {
|
body: {
|
||||||
session_id: sessionId,
|
session_id: sessionId,
|
||||||
messages: currentMessages,
|
user_message: responseMessage,
|
||||||
},
|
},
|
||||||
throwOnError: true,
|
throwOnError: true,
|
||||||
signal: abortControllerRef.current.signal,
|
signal: abortControllerRef.current.signal,
|
||||||
|
|||||||
Reference in New Issue
Block a user