RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #889 from plastic-labs/eugene/dev-1989 

Split `REPRESENTATION_BATCH_MAX_TOKENS` into a "minimum work unit" setting on the producer side and a "maximum LLM tokens" setting on the consumer side
This commit is contained in:
Eugene Eisenstein 2026-07-09 15:12:27 -04:00 committed by GitHub
parent f70add3fea 43d962d160
commit 6d19c46861
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 129 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.env.template
View File

@ -131,7 +131,8 @@ LLM_OPENAI_API_KEY=your-api-key-here
# DERIVER_MAX_INPUT_TOKENS=25000
# DERIVER_MAX_CUSTOM_INSTRUCTIONS_TOKENS=2000
# DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100
# DERIVER_REPRESENTATION_BATCH_MAX_TOKENS=1024
# DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=512 # Min tokens a work unit accumulates before the deriver claims it; 0 disables the gate
# DERIVER_REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=1024 # Max context-window tokens per deriver LLM call
# DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS=1800
# DERIVER_FLUSH_ENABLED=false # Bypass batch token threshold, process work immediately
# DERIVER_MODEL_CONFIG__FALLBACK__MODEL=

3
.gitignore vendored
View File

@ -193,3 +193,6 @@ metrics.jsonl
AGENTS.md
lancedb_data/
grafana-data/
# Claude Code addon stuff
.omc

3
config.toml.example
View File

@ -109,7 +109,8 @@ LOG_OBSERVATIONS = false
MAX_INPUT_TOKENS = 25000
MAX_CUSTOM_INSTRUCTIONS_TOKENS = 2000
WORKING_REPRESENTATION_MAX_OBSERVATIONS = 100
REPRESENTATION_BATCH_MAX_TOKENS = 1024
REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS = 512 # Min tokens a work unit accumulates before the deriver claims it; 0 disables the gate
REPRESENTATION_BATCH_TARGET_INPUT_TOKENS = 1024 # Max context-window tokens per deriver LLM call
REPRESENTATION_BATCH_MAX_AGE_SECONDS = 1800
FLUSH_ENABLED = false # Bypass batch token threshold, process work immediately

3
docs/v3/contributing/configuration.mdx
View File

@ -408,7 +408,8 @@ DERIVER_QUEUE_ERROR_RETENTION_SECONDS=2592000 # 30 days
DERIVER_DEDUPLICATE=true
DERIVER_LOG_OBSERVATIONS=false
DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100
DERIVER_REPRESENTATION_BATCH_MAX_TOKENS=1024
DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=512
DERIVER_REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=1024
DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS=1800
```

2
docs/v3/contributing/troubleshooting.mdx
View File

@ -109,7 +109,7 @@ Messages are stored but no observations, summaries, or representations are being
```bash
DERIVER_WORKERS=4
```
5. **Representation Batch Max** — By default the deriver buffers representation work until a session has enough tokens for that representation, set via `DERIVER_REPRESENTATION_BATCH_MAX_TOKENS`. Sub-threshold tails become eligible after `DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS` (default 1800 seconds), so quiet sessions eventually flush without disabling batching globally. Set the age to `0` for legacy behavior where sub-threshold tails wait indefinitely. See [token batching](/v3/documentation/core-concepts/reasoning#token-batching) for more details
5. **Representation Batching** — By default the deriver buffers representation work until a work unit has accumulated enough tokens, set via `DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS` (`0` disables the accumulation gate). A separate setting, `DERIVER_REPRESENTATION_BATCH_TARGET_INPUT_TOKENS`, caps the conversation window fed to each deriver LLM call when draining a claimed work unit. Sub-threshold tails become eligible after `DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS` (default 1800 seconds), so quiet sessions eventually flush without disabling batching globally. Set the age to `0` for legacy behavior where sub-threshold tails wait indefinitely. See [token batching](/v3/documentation/core-concepts/reasoning#token-batching) for more details
## Alternative Provider Issues

41
src/config.py
View File

@ -861,7 +861,20 @@ class DeriverSettings(HonchoSettings):
int, Field(default=100, gt=0, le=1000)
] = 100
REPRESENTATION_BATCH_MAX_TOKENS: Annotated[
# Minimum tokens a representation work unit must accumulate (summed over
# its own unprocessed messages) before it becomes claimable. Bypassed by
# FLUSH_ENABLED and by REPRESENTATION_BATCH_MAX_AGE_SECONDS age-flushing.
# 0 disables the accumulation gate entirely (equivalent to FLUSH_ENABLED
# for claiming): work units are claimable as soon as anything is pending.
REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS: Annotated[
int,
Field(default=512, ge=0, le=16_384),
] = 512
# Cumulative-token cap on the conversation window (queued messages plus
# interleaved context) fed to a single deriver LLM call when draining a
# claimed work unit. The first unprocessed message is always included,
# even if it alone exceeds the cap.
REPRESENTATION_BATCH_TARGET_INPUT_TOKENS: Annotated[
int,
Field(default=1024, ge=128, le=16_384),
] = 1024
@ -885,11 +898,33 @@ class DeriverSettings(HonchoSettings):
)
return data # pyright: ignore[reportUnknownVariableType]
@model_validator(mode="before")
@classmethod
def _reject_removed_batch_max_tokens(cls, data: Any) -> Any:
"""Fail fast on the removed REPRESENTATION_BATCH_MAX_TOKENS setting.
The old single setting was split into
REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS (claim gate) and
REPRESENTATION_BATCH_TARGET_INPUT_TOKENS (per-LLM-call window cap).
`extra="ignore"` would otherwise silently drop the old key and revert
both roles to defaults — an operator-hostile failure mode for a
batching knob — so reject it loudly instead.
"""
legacy_in_data = isinstance(data, dict) and any(
str(key).upper() == "REPRESENTATION_BATCH_MAX_TOKENS"
for key in cast(dict[str, Any], data)
)
if legacy_in_data or "DERIVER_REPRESENTATION_BATCH_MAX_TOKENS" in os.environ:
raise ValueError(
"REPRESENTATION_BATCH_MAX_TOKENS has been split into REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS (minimum tokens a work unit must accumulate before it is claimed) and REPRESENTATION_BATCH_TARGET_INPUT_TOKENS (token cap on the context window per deriver LLM call). Set those instead."
)
return data # pyright: ignore[reportUnknownVariableType]
@model_validator(mode="after")
def validate_batch_tokens_vs_context_limit(self):
if self.REPRESENTATION_BATCH_MAX_TOKENS > self.MAX_INPUT_TOKENS:
if self.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS > self.MAX_INPUT_TOKENS:
raise ValueError(
f"REPRESENTATION_BATCH_MAX_TOKENS ({self.REPRESENTATION_BATCH_MAX_TOKENS}) cannot exceed max deriver input tokens ({self.MAX_INPUT_TOKENS})"
f"REPRESENTATION_BATCH_TARGET_INPUT_TOKENS ({self.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS}) cannot exceed max deriver input tokens ({self.MAX_INPUT_TOKENS})"
)
return self

2
src/deriver/consumer.py
View File

@ -176,7 +176,7 @@ async def process_representation_batch(
queue_item_message_ids: Message IDs from queue items
hit_batch_token_cap: whether the queue batcher clamped this batch to fit
was_flush_enabled: snapshot of DERIVER.FLUSH_ENABLED at fetch time
batch_max_tokens: DERIVER.REPRESENTATION_BATCH_MAX_TOKENS snapshot
batch_max_tokens: DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS snapshot
"""
if not messages or not messages[0]:
logger.debug("process_representation_batch received no messages")

2
src/deriver/deriver.py
View File

@ -58,7 +58,7 @@ async def process_representation_tasks_batch(
queue_item_message_ids: Message IDs from queue items being processed
hit_batch_token_cap: queue batcher clamped this batch to fit
was_flush_enabled: DERIVER.FLUSH_ENABLED snapshot at batch time
batch_max_tokens: DERIVER.REPRESENTATION_BATCH_MAX_TOKENS snapshot
batch_max_tokens: DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS snapshot
"""
if not messages:
return

20
src/deriver/queue_manager.py
View File

@ -331,16 +331,18 @@ class QueueManager:
"""
Get available work units that aren't being processed.
For representation tasks, only returns work units whose accumulated
tokens reach REPRESENTATION_BATCH_MAX_TOKENS or whose oldest pending
item exceeds REPRESENTATION_BATCH_MAX_AGE_SECONDS, unless
FLUSH_ENABLED is True.
tokens reach REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS or whose
oldest pending item exceeds REPRESENTATION_BATCH_MAX_AGE_SECONDS,
unless FLUSH_ENABLED is True.
Returns a dict mapping work_unit_key to aqs_id.
"""
limit: int = max(0, self.workers - self.get_total_owned_work_units())
if limit == 0:
return {}
batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
work_unit_target_tokens = (
settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS
)
async with tracked_db("get_available_work_units") as db:
representation_prefix = "representation:"
@ -396,11 +398,11 @@ class QueueManager:
)
# Apply batch threshold filter (skip if FLUSH_ENABLED is True)
if not settings.DERIVER.FLUSH_ENABLED and batch_max_tokens > 0:
if not settings.DERIVER.FLUSH_ENABLED and work_unit_target_tokens > 0:
max_age_seconds = settings.DERIVER.REPRESENTATION_BATCH_MAX_AGE_SECONDS
threshold_clause = (
func.coalesce(token_stats_subq.c.total_tokens, 0)
>= batch_max_tokens
>= work_unit_target_tokens
)
if max_age_seconds > 0:
threshold_clause = or_(
@ -426,13 +428,13 @@ class QueueManager:
not settings.DERIVER.FLUSH_ENABLED
and settings.DERIVER.REPRESENTATION_BATCH_MAX_AGE_SECONDS > 0
and work_unit_key.startswith(representation_prefix)
and int(total_tokens or 0) < batch_max_tokens
and int(total_tokens or 0) < work_unit_target_tokens
):
logger.info(
"age-flushing work unit %s (tokens=%s < %s, oldest=%s)",
work_unit_key,
total_tokens or 0,
batch_max_tokens,
work_unit_target_tokens,
oldest_created_at,
)
if not available_units:
@ -816,7 +818,7 @@ class QueueManager:
f"{task_type} tasks are not supported for get_queue_item_batch"
)
batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS
was_flush_enabled = settings.DERIVER.FLUSH_ENABLED
parsed_key = parse_work_unit_key(work_unit_key)
messages_context: list[models.Message] = []

2
src/telemetry/events/representation.py
View File

@ -104,7 +104,7 @@ class RepresentationCompletedEvent(BaseEvent):
# Cap configuration + hit flags ()
batch_max_tokens: int = Field(
default=0,
description="settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS at fetch time",
description="settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS at fetch time",
)
max_input_tokens: int = Field(
default=0, description="settings.DERIVER.MAX_INPUT_TOKENS at call time"

2
tests/bench/calculate_expected_events.py
View File

@ -246,7 +246,7 @@ def calculate_question_events(
# Calculate representation events
# Each unique (session, observed) pair generates one representation event
# (assuming messages fit within REPRESENTATION_BATCH_MAX_TOKENS)
# (assuming messages fit within REPRESENTATION_BATCH_TARGET_INPUT_TOKENS)
# When merge_sessions=True, all messages go into one session
if merge_sessions:
# One merged session = one representation event

24
tests/deriver/test_queue_processing.py
View File

@ -359,7 +359,7 @@ class TestQueueProcessing:
peer = peers[0]
# Create messages with token counts that exceed batch limit
limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
limit = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS
token_counts = [limit // 2, limit // 2, limit // 2]
# Create and save messages to the database first
@ -479,7 +479,7 @@ class TestQueueProcessing:
session, peers = sample_session_with_peers
peer = peers[0]
cap = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
cap = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS
# M1 + M2 sum to exactly the cap; M3 pushes over it. After SQL,
# messages_context = [M1, M2]; the cap is genuinely binding *on the
@ -581,7 +581,7 @@ class TestQueueProcessing:
session, peers = sample_session_with_peers
peer_a = peers[0]
peer_b = peers[1] if len(peers) > 1 else peers[0]
cap = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
cap = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS
# Layout: 4 messages, ordered.
# M1 (peer_a, queue, 200)
@ -744,7 +744,9 @@ class TestQueueProcessing:
qm = QueueManager()
# Mock the token limit to 2000 for this test
with patch.object(settings.DERIVER, "REPRESENTATION_BATCH_MAX_TOKENS", 2000):
with patch.object(
settings.DERIVER, "REPRESENTATION_BATCH_TARGET_INPUT_TOKENS", 2000
):
# Test alice's work unit
alice_work_unit_key = alice_queue_items[0].work_unit_key
alice_aqs = models.ActiveQueueSession(work_unit_key=alice_work_unit_key)
@ -919,7 +921,9 @@ class TestQueueProcessing:
qm = QueueManager()
# Mock the token limit to 1500 for this test
with patch.object(settings.DERIVER, "REPRESENTATION_BATCH_MAX_TOKENS", 1500):
with patch.object(
settings.DERIVER, "REPRESENTATION_BATCH_TARGET_INPUT_TOKENS", 1500
):
# Test alice's work unit
# With per-work-unit anchoring + preceding context:
# Alice starts at message 3, includes preceding message 2 (steve) for context
@ -1133,7 +1137,7 @@ class TestQueueProcessing:
peer = peers[0]
# Create messages where first message exceeds the batch limit
limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
limit = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS
token_counts = [limit + 1000, 100, 200] # First message way over limit
# Create and save messages to the database first
@ -1250,7 +1254,7 @@ class TestQueueProcessing:
peer = peers[0]
# Create messages that test the exact boundary
limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
limit = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS
token_counts = [
limit // 2,
limit // 2,
@ -1381,7 +1385,7 @@ class TestQueueProcessing:
peer = peers[0]
# Create messages with tokens BELOW the threshold
limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
limit = settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS
token_counts = [100, 100, 100] # Total 300, way below 4096
messages: list[models.Message] = []
@ -1652,7 +1656,7 @@ class TestQueueProcessing:
session, peers = sample_session_with_peers
peer = peers[0]
limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
limit = settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS
# Create a single message that exceeds the threshold
message = models.Message(
@ -1758,7 +1762,7 @@ class TestQueueProcessing:
session, peers = sample_session_with_peers
peer = peers[0]
limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS
limit = settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS
# Create messages that sum to exactly the threshold
token_counts = [limit // 2, limit // 2]

54
tests/test_config.py
View File

@ -7,7 +7,8 @@ def _make_deriver_settings(
*,
MAX_INPUT_TOKENS: int = 25000,
MAX_CUSTOM_INSTRUCTIONS_TOKENS: int = 2000,
REPRESENTATION_BATCH_MAX_TOKENS: int = 1024,
REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS: int = 512,
REPRESENTATION_BATCH_TARGET_INPUT_TOKENS: int = 1024,
REPRESENTATION_BATCH_MAX_AGE_SECONDS: int = 1800,
) -> DeriverSettings:
return DeriverSettings(
@ -17,7 +18,8 @@ def _make_deriver_settings(
),
MAX_INPUT_TOKENS=MAX_INPUT_TOKENS,
MAX_CUSTOM_INSTRUCTIONS_TOKENS=MAX_CUSTOM_INSTRUCTIONS_TOKENS,
REPRESENTATION_BATCH_MAX_TOKENS=REPRESENTATION_BATCH_MAX_TOKENS,
REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS,
REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=REPRESENTATION_BATCH_TARGET_INPUT_TOKENS,
REPRESENTATION_BATCH_MAX_AGE_SECONDS=REPRESENTATION_BATCH_MAX_AGE_SECONDS,
)
@ -50,3 +52,51 @@ def test_representation_batch_age_can_be_disabled_with_zero() -> None:
def test_representation_batch_age_rejects_negative_values() -> None:
with pytest.raises(ValueError, match="greater than or equal to 0"):
_make_deriver_settings(REPRESENTATION_BATCH_MAX_AGE_SECONDS=-1)
def test_representation_batch_work_unit_target_can_be_disabled_with_zero() -> None:
settings = _make_deriver_settings(REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=0)
assert settings.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS == 0
def test_representation_batch_work_unit_target_rejects_negative_values() -> None:
with pytest.raises(ValueError, match="greater than or equal to 0"):
_make_deriver_settings(REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=-1)
def test_representation_batch_tokens_can_diverge() -> None:
settings = _make_deriver_settings(
REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=4096,
REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=1024,
)
assert settings.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS == 4096
assert settings.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS == 1024
def test_representation_batch_target_input_cannot_exceed_max_input_tokens() -> None:
with pytest.raises(ValueError, match="cannot exceed max deriver input tokens"):
_make_deriver_settings(
MAX_INPUT_TOKENS=1000,
REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=2048,
)
def test_legacy_representation_batch_max_tokens_is_rejected() -> None:
with pytest.raises(ValueError, match="has been split into"):
DeriverSettings(
MODEL_CONFIG=ConfiguredModelSettings(
model="gpt-5.4-mini",
transport="openai",
),
REPRESENTATION_BATCH_MAX_TOKENS=1024, # pyright: ignore[reportCallIssue]
)
def test_legacy_representation_batch_max_tokens_env_var_is_rejected(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("DERIVER_REPRESENTATION_BATCH_MAX_TOKENS", "1024")
with pytest.raises(ValueError, match="has been split into"):
_make_deriver_settings()