apunkt/plano
1
0
Fork 0
mirror of https://github.com/katanemo/plano.git synced 2026-06-26 15:39:40 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix logs a bit more and refactor code

Browse source
This commit is contained in:
Adil Hafeez 2025-01-27 15:44:41 -08:00
parent a0b13b99ce
commit cc730d8b3d
No known key found for this signature in database
GPG key ID: 9B18EF7691369645
4 changed files with 40 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

7
crates/llm_gateway/src/stream_context.rs
View file

@ -517,8 +517,11 @@ impl HttpContext for StreamContext {
let chat_completions_response: ChatCompletionsResponse = let chat_completions_response: ChatCompletionsResponse =
match serde_json::from_str(body_utf8.as_str()) { match serde_json::from_str(body_utf8.as_str()) {
Ok(de) => de, Ok(de) => de,
Err(_e) => { Err(err) => {
debug!("invalid response: {}", body_utf8); debug!(
"non chat-completion compliant response received err: {}, body: {}",
err, body_utf8
);
return Action::Continue; return Action::Continue;
} }
}; };

42
crates/prompt_gateway/src/context.rs
View file

@ -27,20 +27,36 @@ impl Context for StreamContext {
.get_http_call_response_body(0, body_size) .get_http_call_response_body(0, body_size)
.unwrap_or_default(); .unwrap_or_default();
let http_status = self if let Some(http_status) = self.get_http_call_response_header(":status") {
.get_http_call_response_header(":status") match StatusCode::from_str(http_status.as_str()) {
.unwrap_or(StatusCode::OK.as_str().to_string()); Ok(status_code) => {
if http_status != StatusCode::OK.as_str() { if !status_code.is_success() {
let server_error = ServerError::Upstream { let server_error = ServerError::Upstream {
host: callout_context.upstream_cluster.unwrap(), host: callout_context.upstream_cluster.unwrap(),
path: callout_context.upstream_cluster_path.unwrap(), path: callout_context.upstream_cluster_path.unwrap(),
status: http_status.clone(), status: http_status.clone(),
body: String::from_utf8(body).unwrap(), body: String::from_utf8(body).unwrap(),
}; };
warn!("filter received non 2xx code: {:?}", server_error); warn!("received non 2xx code: {:?}", server_error);
return self.send_server_error(
server_error,
Some(StatusCode::from_str(http_status.as_str()).unwrap()),
);
}
}
Err(_) => {
// invalid status code (status code non numeric)
return self.send_server_error(
ServerError::LogicError(format!("invalid status code: {}", http_status)),
Some(StatusCode::from_str(http_status.as_str()).unwrap()),
);
}
}
} else {
// :status header not found
return self.send_server_error( return self.send_server_error(
server_error, ServerError::LogicError("missing :status header".to_string()),
Some(StatusCode::from_str(http_status.as_str()).unwrap()), Some(StatusCode::BAD_GATEWAY),
); );
} }

4
crates/prompt_gateway/src/http_context.rs
View file

@ -79,7 +79,7 @@ impl HttpContext for StreamContext {
}; };
debug!( debug!(
"developer => archgw: {}", "developer => archgw request body: {}",
String::from_utf8_lossy(&body_bytes) String::from_utf8_lossy(&body_bytes)
); );
@ -152,7 +152,7 @@ impl HttpContext for StreamContext {
} }
}; };
debug!("archgw => archfc: {}", json_data); debug!("archgw => modelserver request body: {}", json_data);
let mut headers = vec![ let mut headers = vec![
(ARCH_UPSTREAM_HOST_HEADER, MODEL_SERVER_NAME), (ARCH_UPSTREAM_HOST_HEADER, MODEL_SERVER_NAME),

8
crates/prompt_gateway/src/stream_context.rs
View file

@ -125,13 +125,13 @@ impl StreamContext {
mut callout_context: StreamCallContext, mut callout_context: StreamCallContext,
) { ) {
let body_str = String::from_utf8(body).unwrap(); let body_str = String::from_utf8(body).unwrap();
debug!("archgw <= archfc response: {}", body_str); debug!("archgw <= modelserver response body: {}", body_str);
let model_server_response: ModelServerResponse = match serde_json::from_str(&body_str) { let model_server_response: ModelServerResponse = match serde_json::from_str(&body_str) {
Ok(arch_fc_response) => arch_fc_response, Ok(arch_fc_response) => arch_fc_response,
Err(e) => { Err(e) => {
warn!( warn!(
"error deserializing archfc response: {}, body: {}", "error deserializing modelserver response: {}, body: {}",
e, body_str e, body_str
); );
return self.send_server_error(ServerError::Deserialization(e), None); return self.send_server_error(ServerError::Deserialization(e), None);
@ -141,7 +141,7 @@ impl StreamContext {
let arch_fc_response = match model_server_response { let arch_fc_response = match model_server_response {
ModelServerResponse::ChatCompletionsResponse(response) => response, ModelServerResponse::ChatCompletionsResponse(response) => response,
ModelServerResponse::ModelServerErrorResponse(response) => { ModelServerResponse::ModelServerErrorResponse(response) => {
debug!("archgw <= archfc error response: {}", response.result); debug!("archgw <= modelserver error response: {}", response.result);
if response.result == "No intent matched" { if response.result == "No intent matched" {
if let Some(default_prompt_target) = self if let Some(default_prompt_target) = self
.prompt_targets .prompt_targets
@ -344,7 +344,7 @@ impl StreamContext {
); );
debug!( debug!(
"archgw => api call, endpoint: {}{}, body: {}", "archgw => developer api call endpoint: {}, path: {}, body: {}",
endpoint.name.as_str(), endpoint.name.as_str(),
path, path,
tool_params_json_str tool_params_json_str