style: Refactored some imports and LLM hints
This commit is contained in:
4 files changed
+30
-30
No files matched your search
+11
-12
@@ -597,7 +597,7 @@ pub async fn call_chat_completions_streaming(
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Streaming transport for graph `llm` nodes: accumulates the reply and the
|
/// Streaming transport for graph `llm` nodes: accumulates the reply and the
|
||||||
/// transport itself renders nothing — no spinner, delta, or reasoning
|
/// transport itself renders nothing; no spinner, delta, or reasoning
|
||||||
/// (tool-call rendering inside `eval_tool_calls` is unchanged from the
|
/// (tool-call rendering inside `eval_tool_calls` is unchanged from the
|
||||||
/// non-streaming path). Because chunks arrive as they are generated, the
|
/// non-streaming path). Because chunks arrive as they are generated, the
|
||||||
/// reqwest `read_timeout` bounds only stalls between chunks; the caller's
|
/// reqwest `read_timeout` bounds only stalls between chunks; the caller's
|
||||||
@@ -605,7 +605,7 @@ pub async fn call_chat_completions_streaming(
|
|||||||
/// transport is kept deliberately: the handler's call-loop detection is off
|
/// transport is kept deliberately: the handler's call-loop detection is off
|
||||||
/// (that path never had it), and a model the catalog marks `no_stream`
|
/// (that path never had it), and a model the catalog marks `no_stream`
|
||||||
/// takes the non-streaming request, raced against the abort. An abort
|
/// takes the non-streaming request, raced against the abort. An abort
|
||||||
/// observed after the provider call fails outright — partial output is
|
/// observed after the provider call fails outright. Partial output is
|
||||||
/// never returned as success.
|
/// never returned as success.
|
||||||
pub async fn call_chat_completions_streaming_quiet(
|
pub async fn call_chat_completions_streaming_quiet(
|
||||||
input: &Input,
|
input: &Input,
|
||||||
@@ -900,8 +900,10 @@ mod tests {
|
|||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
use super::super::access_token::{is_rejected, set_access_token};
|
use super::super::access_token::{is_rejected, set_access_token};
|
||||||
|
use crate::config::{AppState, WorkingMode};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
|
use tokio::time::Instant;
|
||||||
|
|
||||||
fn catch_error_message(data: &Value, status: u16) -> String {
|
fn catch_error_message(data: &Value, status: u16) -> String {
|
||||||
catch_error(data, status).unwrap_err().to_string()
|
catch_error(data, status).unwrap_err().to_string()
|
||||||
@@ -1189,10 +1191,7 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn quiet_ctx() -> RequestContext {
|
fn quiet_ctx() -> RequestContext {
|
||||||
RequestContext::new(
|
RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd)
|
||||||
Arc::new(crate::config::AppState::test_default()),
|
|
||||||
crate::config::WorkingMode::Cmd,
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 600s of virtual generation completes and yields the same shape the
|
/// 600s of virtual generation completes and yields the same shape the
|
||||||
@@ -1207,7 +1206,7 @@ mod tests {
|
|||||||
let mut ctx = quiet_ctx();
|
let mut ctx = quiet_ctx();
|
||||||
let input = Input::from_str(&ctx, "hi", None).unwrap();
|
let input = Input::from_str(&ctx, "hi", None).unwrap();
|
||||||
let client = paced_client(false);
|
let client = paced_client(false);
|
||||||
let started = tokio::time::Instant::now();
|
let started = Instant::now();
|
||||||
|
|
||||||
let (text, results) =
|
let (text, results) =
|
||||||
call_chat_completions_streaming_quiet(&input, &client, &mut ctx, create_abort_signal())
|
call_chat_completions_streaming_quiet(&input, &client, &mut ctx, create_abort_signal())
|
||||||
@@ -1236,7 +1235,7 @@ mod tests {
|
|||||||
tokio::time::sleep(PACED_GAP * PACED_DELTAS as u32 + Duration::from_secs(30)).await;
|
tokio::time::sleep(PACED_GAP * PACED_DELTAS as u32 + Duration::from_secs(30)).await;
|
||||||
trigger.set_ctrlc();
|
trigger.set_ctrlc();
|
||||||
});
|
});
|
||||||
let started = tokio::time::Instant::now();
|
let started = Instant::now();
|
||||||
|
|
||||||
let err = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, abort)
|
let err = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, abort)
|
||||||
.await
|
.await
|
||||||
@@ -1312,7 +1311,7 @@ mod tests {
|
|||||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||||
trigger.set_ctrlc();
|
trigger.set_ctrlc();
|
||||||
});
|
});
|
||||||
let started = tokio::time::Instant::now();
|
let started = Instant::now();
|
||||||
|
|
||||||
let err = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, abort)
|
let err = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, abort)
|
||||||
.await
|
.await
|
||||||
@@ -1326,14 +1325,14 @@ mod tests {
|
|||||||
/// transport still streams with it off.
|
/// transport still streams with it off.
|
||||||
#[tokio::test(start_paused = true)]
|
#[tokio::test(start_paused = true)]
|
||||||
async fn quiet_transport_ignores_the_user_stream_toggle() {
|
async fn quiet_transport_ignores_the_user_stream_toggle() {
|
||||||
let app = crate::config::AppState {
|
let app = AppState {
|
||||||
config: Arc::new(AppConfig {
|
config: Arc::new(AppConfig {
|
||||||
stream: false,
|
stream: false,
|
||||||
..AppConfig::default()
|
..AppConfig::default()
|
||||||
}),
|
}),
|
||||||
..crate::config::AppState::test_default()
|
..AppState::test_default()
|
||||||
};
|
};
|
||||||
let mut ctx = RequestContext::new(Arc::new(app), crate::config::WorkingMode::Cmd);
|
let mut ctx = RequestContext::new(Arc::new(app), WorkingMode::Cmd);
|
||||||
let input = Input::from_str(&ctx, "hi", None).unwrap();
|
let input = Input::from_str(&ctx, "hi", None).unwrap();
|
||||||
assert!(!input.stream(), "fixture must have streaming disabled");
|
assert!(!input.stream(), "fixture must have streaming disabled");
|
||||||
let client = paced_client(false);
|
let client = paced_client(false);
|
||||||
|
|||||||
+3
-1
@@ -127,7 +127,7 @@ async fn run(
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Drives the retry loop, then extraction, then state updates. Extraction
|
/// Drives the retry loop, then extraction, then state updates. Extraction
|
||||||
/// stays outside the retry loop: by the time it runs the agent itself
|
/// stays outside the retry loop: by the time it runs, the agent itself
|
||||||
/// succeeded, so an extraction failure is never retried.
|
/// succeeded, so an extraction failure is never retried.
|
||||||
async fn attempt_and_extract(
|
async fn attempt_and_extract(
|
||||||
node_id: &str,
|
node_id: &str,
|
||||||
@@ -249,6 +249,7 @@ async fn retry_transient(
|
|||||||
parent_ctx.peer_registry = Some(Arc::clone(registry));
|
parent_ctx.peer_registry = Some(Arc::clone(registry));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
match run_attempt(parent_ctx).await {
|
match run_attempt(parent_ctx).await {
|
||||||
Ok(out) => return Ok(out),
|
Ok(out) => return Ok(out),
|
||||||
Err(e) if is_transient_error(&e) && attempt < node.max_attempts => {
|
Err(e) if is_transient_error(&e) && attempt < node.max_attempts => {
|
||||||
@@ -260,6 +261,7 @@ async fn retry_transient(
|
|||||||
Err(e) => return Err(e),
|
Err(e) => return Err(e),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
Err(last_err.unwrap_or_else(|| anyhow!("agent node exhausted retries")))
|
Err(last_err.unwrap_or_else(|| anyhow!("agent node exhausted retries")))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+12
-13
@@ -16,6 +16,7 @@ pub mod types;
|
|||||||
pub mod user_interaction;
|
pub mod user_interaction;
|
||||||
pub mod validator;
|
pub mod validator;
|
||||||
|
|
||||||
|
use crate::client::ApiStatusError;
|
||||||
use anyhow::Error;
|
use anyhow::Error;
|
||||||
pub use dispatch::{
|
pub use dispatch::{
|
||||||
active_agent_graph_name, run_active_agent_graph, run_active_agent_graph_with_inputs,
|
active_agent_graph_name, run_active_agent_graph, run_active_agent_graph_with_inputs,
|
||||||
@@ -45,10 +46,7 @@ pub(crate) fn is_transient_error(err: &Error) -> bool {
|
|||||||
}) {
|
}) {
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
if let Some(api) = err
|
if let Some(api) = err.chain().find_map(|c| c.downcast_ref::<ApiStatusError>()) {
|
||||||
.chain()
|
|
||||||
.find_map(|c| c.downcast_ref::<crate::client::ApiStatusError>())
|
|
||||||
{
|
|
||||||
return is_transient_status(api.status);
|
return is_transient_status(api.status);
|
||||||
}
|
}
|
||||||
let s = format!("{err:#}").to_lowercase();
|
let s = format!("{err:#}").to_lowercase();
|
||||||
@@ -67,8 +65,8 @@ pub(crate) fn is_transient_error(err: &Error) -> bool {
|
|||||||
|
|
||||||
/// Transient HTTP statuses worth a from-scratch retry: 429 (rate limit),
|
/// Transient HTTP statuses worth a from-scratch retry: 429 (rate limit),
|
||||||
/// 500/502/503/504 (upstream hiccups), 529 (Anthropic overloaded).
|
/// 500/502/503/504 (upstream hiccups), 529 (Anthropic overloaded).
|
||||||
/// Deterministic statuses — 4xx other than 429 (including 408) and
|
/// Deterministic statuses (4xx other than 429 (including 408) and
|
||||||
/// 501/505-class capability errors — are excluded.
|
/// 501/505-class capability errors) are excluded.
|
||||||
fn is_transient_status(status: u16) -> bool {
|
fn is_transient_status(status: u16) -> bool {
|
||||||
matches!(status, 429 | 500 | 502 | 503 | 504 | 529)
|
matches!(status, 429 | 500 | 502 | 503 | 504 | 529)
|
||||||
}
|
}
|
||||||
@@ -89,6 +87,7 @@ pub(in crate::graph) fn type_name(value: &Value) -> &'static str {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::client;
|
||||||
use anyhow::anyhow;
|
use anyhow::anyhow;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -151,7 +150,7 @@ mod tests {
|
|||||||
"message": "Too many requests, please slow down"
|
"message": "Too many requests, please slow down"
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
let err = crate::client::catch_error(&data, 429)
|
let err = client::catch_error(&data, 429)
|
||||||
.unwrap_err()
|
.unwrap_err()
|
||||||
.context("Failed to call chat-completions api")
|
.context("Failed to call chat-completions api")
|
||||||
.context("Agent 'domain-reviewer' failed");
|
.context("Agent 'domain-reviewer' failed");
|
||||||
@@ -166,7 +165,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn is_transient_error_typed_status_overrides_message_text() {
|
fn is_transient_error_typed_status_overrides_message_text() {
|
||||||
let err = crate::client::catch_error(
|
let err = client::catch_error(
|
||||||
&serde_json::json!({ "message": "rate limit reached — connection reset" }),
|
&serde_json::json!({ "message": "rate limit reached — connection reset" }),
|
||||||
400,
|
400,
|
||||||
)
|
)
|
||||||
@@ -176,7 +175,7 @@ mod tests {
|
|||||||
"typed 400 must not be retried even when the body matches string anchors"
|
"typed 400 must not be retried even when the body matches string anchors"
|
||||||
);
|
);
|
||||||
|
|
||||||
let err = crate::client::catch_error(&serde_json::json!({ "message": "x" }), 503)
|
let err = client::catch_error(&serde_json::json!({ "message": "x" }), 503)
|
||||||
.unwrap_err()
|
.unwrap_err()
|
||||||
.context("Agent 'x' failed");
|
.context("Agent 'x' failed");
|
||||||
assert!(is_transient_error(&err));
|
assert!(is_transient_error(&err));
|
||||||
@@ -186,14 +185,14 @@ mod tests {
|
|||||||
fn is_transient_error_typed_status_table() {
|
fn is_transient_error_typed_status_table() {
|
||||||
let data = serde_json::json!({ "message": "x" });
|
let data = serde_json::json!({ "message": "x" });
|
||||||
for status in [429, 500, 502, 503, 504, 529] {
|
for status in [429, 500, 502, 503, 504, 529] {
|
||||||
let err = crate::client::catch_error(&data, status).unwrap_err();
|
let err = client::catch_error(&data, status).unwrap_err();
|
||||||
assert!(
|
assert!(
|
||||||
is_transient_error(&err),
|
is_transient_error(&err),
|
||||||
"status {status} should be transient"
|
"status {status} should be transient"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
for status in [400, 401, 403, 404, 408, 422, 501, 505] {
|
for status in [400, 401, 403, 404, 408, 422, 501, 505] {
|
||||||
let err = crate::client::catch_error(&data, status).unwrap_err();
|
let err = client::catch_error(&data, status).unwrap_err();
|
||||||
assert!(
|
assert!(
|
||||||
!is_transient_error(&err),
|
!is_transient_error(&err),
|
||||||
"status {status} should not be transient"
|
"status {status} should not be transient"
|
||||||
@@ -261,7 +260,7 @@ mod tests {
|
|||||||
.read_timeout(Duration::from_millis(50))
|
.read_timeout(Duration::from_millis(50))
|
||||||
.build()
|
.build()
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let err = crate::client::sse_stream(client.get(format!("http://{addr}/")), |_| Ok(false))
|
let err = client::sse_stream(client.get(format!("http://{addr}/")), |_| Ok(false))
|
||||||
.await
|
.await
|
||||||
.expect_err("a body that never arrives must trip the read timeout");
|
.expect_err("a body that never arrives must trip the read timeout");
|
||||||
server.abort();
|
server.abort();
|
||||||
@@ -308,7 +307,7 @@ mod tests {
|
|||||||
.build()
|
.build()
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let mut received = 0usize;
|
let mut received = 0usize;
|
||||||
crate::client::sse_stream(client.get(format!("http://{addr}/")), |message| {
|
client::sse_stream(client.get(format!("http://{addr}/")), |message| {
|
||||||
if message.data == "[DONE]" {
|
if message.data == "[DONE]" {
|
||||||
return Ok(true);
|
return Ok(true);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ use crate::config;
|
|||||||
use crate::config::{Agent, AppConfig, paths};
|
use crate::config::{Agent, AppConfig, paths};
|
||||||
use crate::rag::{GraphRagConfig, RagData};
|
use crate::rag::{GraphRagConfig, RagData};
|
||||||
use anyhow::{Result, bail};
|
use anyhow::{Result, bail};
|
||||||
use std::collections::{BTreeMap, HashSet, VecDeque};
|
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use tokio::sync::Semaphore;
|
use tokio::sync::Semaphore;
|
||||||
@@ -451,7 +451,7 @@ impl GraphValidator {
|
|||||||
format!(
|
format!(
|
||||||
"{kind} node's `output_schema` declares no (non-empty) object `properties`; \
|
"{kind} node's `output_schema` declares no (non-empty) object `properties`; \
|
||||||
the engine auto-merges ONLY keys declared under `properties` (a bare \
|
the engine auto-merges ONLY keys declared under `properties` (a bare \
|
||||||
object schema merges nothing) — declare the keys under \
|
object schema merges nothing). Declare the keys under \
|
||||||
`properties` or lift them via `state_updates`"
|
`properties` or lift them via `state_updates`"
|
||||||
),
|
),
|
||||||
));
|
));
|
||||||
@@ -1297,7 +1297,7 @@ fn primary_templated_fields(node: &Node) -> Vec<String> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn node_state_updates_map(node: &Node) -> Option<&std::collections::HashMap<String, String>> {
|
fn node_state_updates_map(node: &Node) -> Option<&HashMap<String, String>> {
|
||||||
match &node.node_type {
|
match &node.node_type {
|
||||||
NodeType::Llm(n) => n.state_updates.as_ref(),
|
NodeType::Llm(n) => n.state_updates.as_ref(),
|
||||||
NodeType::Agent(n) => n.state_updates.as_ref(),
|
NodeType::Agent(n) => n.state_updates.as_ref(),
|
||||||
@@ -1310,7 +1310,7 @@ fn node_state_updates_map(node: &Node) -> Option<&std::collections::HashMap<Stri
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn lifts_via_state_updates(updates: &Option<std::collections::HashMap<String, String>>) -> bool {
|
fn lifts_via_state_updates(updates: &Option<HashMap<String, String>>) -> bool {
|
||||||
updates.as_ref().is_some_and(|m| !m.is_empty())
|
updates.as_ref().is_some_and(|m| !m.is_empty())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user