Agent: drop the oldest turns when the context is too large; answer by ending the turn with text
This commit is contained in:
+121
-4
@@ -13,6 +13,7 @@ use wstd::time::{Duration, Instant};
|
||||
const SYSTEM: &str = include_str!("prompt.md");
|
||||
const CONFIG: &str = "/data/config.json";
|
||||
const HISTORY: &str = "/data/history.jsonl";
|
||||
const HISTORY_SKIP: &str = "/data/history.skip";
|
||||
const TURN: &str = "/data/turn.json";
|
||||
|
||||
fn main() {
|
||||
@@ -118,6 +119,7 @@ struct TurnState {
|
||||
struct Agent {
|
||||
inbox: Inbox,
|
||||
history: Vec<Entry>,
|
||||
history_skip: usize,
|
||||
pending: TurnState,
|
||||
shutdown: bool,
|
||||
}
|
||||
@@ -129,6 +131,10 @@ impl Agent {
|
||||
.lines()
|
||||
.filter_map(|line| serde_json::from_str(line).ok())
|
||||
.collect();
|
||||
let history_skip = std::fs::read_to_string(HISTORY_SKIP)
|
||||
.ok()
|
||||
.and_then(|skip| skip.trim().parse::<usize>().ok())
|
||||
.unwrap_or(0);
|
||||
let pending = std::fs::read_to_string(TURN)
|
||||
.ok()
|
||||
.and_then(|contents| {
|
||||
@@ -150,6 +156,7 @@ impl Agent {
|
||||
let agent = Self {
|
||||
inbox: Inbox::start(),
|
||||
history,
|
||||
history_skip,
|
||||
pending,
|
||||
shutdown: false,
|
||||
};
|
||||
@@ -241,9 +248,9 @@ impl Agent {
|
||||
text
|
||||
}
|
||||
|
||||
fn has_turn_user(&self, message: &Message) -> bool {
|
||||
fn turn_user_index(&self, message: &Message) -> Option<usize> {
|
||||
let marker = format!("Message {} from {}:\n", message.id, message.from);
|
||||
self.history.iter().any(|entry| match entry {
|
||||
self.history.iter().rposition(|entry| match entry {
|
||||
Entry::User { text } => text.contains(&marker),
|
||||
_ => false,
|
||||
})
|
||||
@@ -285,7 +292,7 @@ impl Agent {
|
||||
let Some(message) = self.pending.active.clone() else {
|
||||
return;
|
||||
};
|
||||
if !self.has_turn_user(&message) {
|
||||
if self.turn_user_index(&message).is_none() {
|
||||
let text = self.turn_text(&message);
|
||||
self.add_history(Entry::User { text });
|
||||
self.pending.notes.clear();
|
||||
@@ -337,7 +344,20 @@ impl Agent {
|
||||
self.preserve_closed_messages();
|
||||
return;
|
||||
}
|
||||
let response = llm::complete(cfg, SYSTEM, &tools, &self.history).await;
|
||||
let current_start = self
|
||||
.turn_user_index(message)
|
||||
.or_else(|| {
|
||||
self.history
|
||||
.iter()
|
||||
.rposition(|entry| matches!(entry, Entry::User { .. }))
|
||||
})
|
||||
.unwrap_or(0);
|
||||
let start = history_start(&self.history, self.history_skip, current_start);
|
||||
if start != self.history_skip {
|
||||
self.history_skip = start;
|
||||
let _ = std::fs::write(HISTORY_SKIP, start.to_string());
|
||||
}
|
||||
let response = llm::complete(cfg, SYSTEM, &tools, &self.history[start..]).await;
|
||||
if self.inbox.is_closed() {
|
||||
self.preserve_closed_messages();
|
||||
return;
|
||||
@@ -345,6 +365,13 @@ impl Agent {
|
||||
let entries = match response {
|
||||
Ok(entries) => entries,
|
||||
Err(e) => {
|
||||
if is_too_large(&e) {
|
||||
if let Some(next) = next_user_start(&self.history, start, current_start) {
|
||||
self.history_skip = next;
|
||||
let _ = std::fs::write(HISTORY_SKIP, next.to_string());
|
||||
continue;
|
||||
}
|
||||
}
|
||||
self.add_history(Entry::Assistant {
|
||||
text: format!("(inference failed: {e})"),
|
||||
phase: None,
|
||||
@@ -574,3 +601,93 @@ fn load_config() -> Result<Config, String> {
|
||||
fn starts_turn(message: &Message) -> bool {
|
||||
message.kind() == "text" && message.reply_to.is_none()
|
||||
}
|
||||
|
||||
fn is_too_large(error: &str) -> bool {
|
||||
let Some((status, body)) = error
|
||||
.strip_prefix("HTTP ")
|
||||
.and_then(|error| error.split_once(':'))
|
||||
else {
|
||||
return false;
|
||||
};
|
||||
let status = status.split_whitespace().next().unwrap_or("");
|
||||
if !matches!(status, "400" | "413") {
|
||||
return false;
|
||||
}
|
||||
let body = body.to_ascii_lowercase();
|
||||
body.contains("context_length_exceeded")
|
||||
|| body.contains("maximum context length")
|
||||
|| body.contains("context length exceeded")
|
||||
|| body.contains("too large")
|
||||
|| (body.contains("too long")
|
||||
&& ["context", "input", "prompt"]
|
||||
.iter()
|
||||
.any(|term| body.contains(term)))
|
||||
}
|
||||
|
||||
fn history_start(history: &[Entry], skip: usize, current: usize) -> usize {
|
||||
let Some(last) = history.len().checked_sub(1) else {
|
||||
return 0;
|
||||
};
|
||||
let current = current.min(last);
|
||||
(skip..=current)
|
||||
.find(|&index| matches!(&history[index], Entry::User { .. }))
|
||||
.unwrap_or(current)
|
||||
}
|
||||
|
||||
fn next_user_start(history: &[Entry], start: usize, current: usize) -> Option<usize> {
|
||||
let end = current.min(history.len().checked_sub(1)?);
|
||||
(start.saturating_add(1)..=end).find(|&index| matches!(&history[index], Entry::User { .. }))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn detects_context_size_errors_only_for_http_400_or_413() {
|
||||
assert!(is_too_large("HTTP 400: context_length_exceeded"));
|
||||
assert!(is_too_large("HTTP 413: prompt is too long"));
|
||||
assert!(is_too_large("HTTP 400: maximum context length"));
|
||||
assert!(is_too_large("HTTP 400: request is too large"));
|
||||
assert!(!is_too_large("HTTP 500: prompt is too long"));
|
||||
assert!(!is_too_large("HTTP 400: invalid token"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn history_skip_advances_by_whole_user_turns() {
|
||||
let history = vec![
|
||||
Entry::User {
|
||||
text: "first".into(),
|
||||
},
|
||||
Entry::ToolCall {
|
||||
id: "a".into(),
|
||||
name: "tool".into(),
|
||||
arguments: "{}".into(),
|
||||
},
|
||||
Entry::ToolResult {
|
||||
id: "a".into(),
|
||||
output: "done".into(),
|
||||
},
|
||||
Entry::User {
|
||||
text: "second".into(),
|
||||
},
|
||||
Entry::ToolCall {
|
||||
id: "b".into(),
|
||||
name: "tool".into(),
|
||||
arguments: "{}".into(),
|
||||
},
|
||||
Entry::ToolResult {
|
||||
id: "b".into(),
|
||||
output: "done".into(),
|
||||
},
|
||||
Entry::User {
|
||||
text: "current".into(),
|
||||
},
|
||||
];
|
||||
assert_eq!(next_user_start(&history, 0, 6), Some(3));
|
||||
assert_eq!(history_start(&history, 1, 6), 3);
|
||||
assert_eq!(next_user_start(&history, 3, 6), Some(6));
|
||||
assert_eq!(history_start(&history, usize::MAX, 6), 6);
|
||||
assert_eq!(next_user_start(&history, 6, 6), None);
|
||||
}
|
||||
}
|
||||
|
||||
+3
-1
@@ -6,6 +6,8 @@ Sausage is a system of actors. Each actor is a WASM program that sends JSON mess
|
||||
|
||||
Actors read one JSON message per line. Each message has `id`, `from`, optional `reply_to`, and `body`. The broker routes a message with `to`. Without `to`, the broker publishes it to subscribers. A reply is a new message with `reply_to` set to the original ID.
|
||||
|
||||
Answer the person by ending your turn with text. Do not send them messages with `send_message` unless they asked for a later notification.
|
||||
|
||||
Use `sausage-actor = "0.1"` in a Rust actor's `Cargo.toml`.
|
||||
|
||||
A simple actor reads messages with `messages()` and replies with `reply()`:
|
||||
@@ -76,6 +78,6 @@ Your private state lives under `/data`. Your public files live under `/public`.
|
||||
|
||||
The host will provide `/actor/<id>/` with `actor.wasm`, `actor.json`, `source/`, `public/`, `status.json`, and `stderr.log`. Each `versions/<n>/` contains `actor.wasm`, `source/`, and `meta.json`.
|
||||
|
||||
Keep state in `/data` so it survives restarts and upgrades. Read the old data format when you upgrade an actor.
|
||||
Keep state in `/data` so it survives restarts and upgrades. Read the old data format when you upgrade an actor. Large contexts may omit the oldest whole turns from requests, so you may not see them even though `/data/history.jsonl` stays complete.
|
||||
|
||||
Your source lives at `/actor/agent/source`. To change yourself, copy it to `/public/drafts/agent`, build it with `cargo`, and send an `upgrade` request to `broker`. The upgrade restarts you. Your turn resumes from `/data/turn.json`.
|
||||
|
||||
Reference in New Issue
Block a user