fix(llm): keep state-changing tool calls in the order the model asked for
Review on #1100 caught that the previous commit parallelised every tool call in a turn, mutations included. `ctx.db_lock` makes the mutating handlers take turns but does not decide whose turn comes first, so two writes from one assistant turn could apply in an order the model did not request — `tool_results` stayed ordered while the database did not. Reads may still overlap freely, because they change nothing. So the loop now splits the turn: anything named in `mutating_tools` runs one at a time in request order, everything else runs concurrently alongside it. `mutating_tools=None` is the default and means "assume every tool mutates", which is the fully sequential behaviour that existed before this branch. A caller opts in by naming its mutating set, so no existing caller changes behaviour by being left alone. `MUTATING_TOOL_NAMES` lives in agent_tools next to the dispatch table and lists exactly the five names routing to a handler that takes `ctx.db_lock`: create_observations, create_observations_deductive, create_observations_inductive, update_peer_card, delete_observations. `extract_preferences` and `finish_consolidation` are deliberately absent — both only return text telling the model what to call next. Dialectic passes that set rather than a "these are all reads" boolean. Its loadout is reads today, but DIALECTIC_TOOLS already carries a commented-out create_observations_deductive, so a flag would have gone quietly wrong the day someone uncommented it. The parameter is threaded through honcho_llm_call instead of imported inside src/llm, which keeps the llm layer free of tool semantics and avoids the existing agent_tools <- dreamer.specialists cycle. Two tests added: mutating calls are not reordered even when a later call finishes first, and the default schedule is fully sequential.
This commit is contained in:
parent
5535415e89
commit
083ef01aab
|
|
@ -35,6 +35,7 @@ from src.telemetry.prometheus.metrics import DialecticComponents, TokenTypes
|
|||
from src.utils.agent_tools import (
|
||||
DIALECTIC_TOOLS,
|
||||
DIALECTIC_TOOLS_MINIMAL,
|
||||
MUTATING_TOOL_NAMES,
|
||||
create_tool_executor,
|
||||
search_memory,
|
||||
)
|
||||
|
|
@ -508,6 +509,11 @@ class DialecticAgent:
|
|||
tool_choice=level_settings.TOOL_CHOICE,
|
||||
tool_executor=tool_executor,
|
||||
max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS,
|
||||
# Dialectic's loadout is reads today, but say so explicitly rather
|
||||
# than relying on that: the loop keeps any state-changing call
|
||||
# in the order the model asked for, so adding a write tool here
|
||||
# later stays correct without anyone remembering this.
|
||||
mutating_tools=MUTATING_TOOL_NAMES,
|
||||
messages=self.messages,
|
||||
max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS,
|
||||
trace_name="dialectic_chat",
|
||||
|
|
@ -584,6 +590,11 @@ class DialecticAgent:
|
|||
tool_choice=level_settings.TOOL_CHOICE,
|
||||
tool_executor=tool_executor,
|
||||
max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS,
|
||||
# Dialectic's loadout is reads today, but say so explicitly rather
|
||||
# than relying on that: the loop keeps any state-changing call
|
||||
# in the order the model asked for, so adding a write tool here
|
||||
# later stays correct without anyone remembering this.
|
||||
mutating_tools=MUTATING_TOOL_NAMES,
|
||||
messages=self.messages,
|
||||
max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS,
|
||||
trace_name="dialectic_chat",
|
||||
|
|
|
|||
|
|
@ -68,6 +68,7 @@ async def honcho_llm_call(
|
|||
tool_choice: str | dict[str, Any] | None = None,
|
||||
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
|
||||
max_tool_iterations: int = 10,
|
||||
mutating_tools: frozenset[str] | None = None,
|
||||
messages: list[dict[str, Any]] | None = None,
|
||||
max_input_tokens: int | None = None,
|
||||
trace_name: str | None = None,
|
||||
|
|
@ -97,6 +98,7 @@ async def honcho_llm_call(
|
|||
tool_choice: str | dict[str, Any] | None = None,
|
||||
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
|
||||
max_tool_iterations: int = 10,
|
||||
mutating_tools: frozenset[str] | None = None,
|
||||
messages: list[dict[str, Any]] | None = None,
|
||||
max_input_tokens: int | None = None,
|
||||
trace_name: str | None = None,
|
||||
|
|
@ -126,6 +128,7 @@ async def honcho_llm_call(
|
|||
tool_choice: str | dict[str, Any] | None = None,
|
||||
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
|
||||
max_tool_iterations: int = 10,
|
||||
mutating_tools: frozenset[str] | None = None,
|
||||
messages: list[dict[str, Any]] | None = None,
|
||||
max_input_tokens: int | None = None,
|
||||
trace_name: str | None = None,
|
||||
|
|
@ -154,6 +157,7 @@ async def honcho_llm_call(
|
|||
tool_choice: str | dict[str, Any] | None = None,
|
||||
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
|
||||
max_tool_iterations: int = 10,
|
||||
mutating_tools: frozenset[str] | None = None,
|
||||
messages: list[dict[str, Any]] | None = None,
|
||||
max_input_tokens: int | None = None,
|
||||
trace_name: str | None = None,
|
||||
|
|
@ -450,6 +454,7 @@ async def honcho_llm_call(
|
|||
tool_choice=tool_choice,
|
||||
tool_executor=tool_executor,
|
||||
max_tool_iterations=max_tool_iterations,
|
||||
mutating_tools=mutating_tools,
|
||||
response_model=response_model,
|
||||
json_mode=json_mode,
|
||||
temperature=temperature,
|
||||
|
|
|
|||
|
|
@ -353,6 +353,7 @@ async def execute_tool_loop(
|
|||
tool_choice: str | dict[str, Any] | None,
|
||||
tool_executor: Callable[[str, dict[str, Any]], Any],
|
||||
max_tool_iterations: int,
|
||||
mutating_tools: frozenset[str] | None = None,
|
||||
response_model: type[BaseModel] | None,
|
||||
json_mode: bool,
|
||||
temperature: float | None,
|
||||
|
|
@ -649,17 +650,37 @@ async def execute_tool_loop(
|
|||
None,
|
||||
)
|
||||
|
||||
# gather preserves argument order, so tool_results and
|
||||
# all_tool_calls stay in the order the model asked for them.
|
||||
outcomes = await asyncio.gather(
|
||||
*(
|
||||
asyncio.create_task(run_tool_call(seq, tool_call))
|
||||
for seq, tool_call in enumerate(response.tool_calls_made)
|
||||
)
|
||||
)
|
||||
# Reads may overlap freely; state-changing calls may not, because
|
||||
# `ctx.db_lock` decides that they take turns but not which turn each
|
||||
# one gets. Anything named in `mutating_tools` therefore runs in the
|
||||
# order the model asked for, while the rest run concurrently
|
||||
# alongside it. `mutating_tools=None` means "assume every tool
|
||||
# mutates", so a caller that has not opted in keeps the fully
|
||||
# sequential behaviour it had before.
|
||||
calls = list(enumerate(response.tool_calls_made))
|
||||
|
||||
def _must_stay_ordered(tool_call: dict[str, Any]) -> bool:
|
||||
return mutating_tools is None or tool_call["name"] in mutating_tools
|
||||
|
||||
# create_task even for the ordered ones: each still needs its own
|
||||
# context copy for the telemetry ContextVars, and awaiting them one
|
||||
# at a time is what preserves the order.
|
||||
concurrent = {
|
||||
seq: asyncio.create_task(run_tool_call(seq, tool_call))
|
||||
for seq, tool_call in calls
|
||||
if not _must_stay_ordered(tool_call)
|
||||
}
|
||||
|
||||
outcomes: dict[int, tuple[dict[str, Any], dict[str, Any] | None]] = {}
|
||||
for seq, tool_call in calls:
|
||||
if _must_stay_ordered(tool_call):
|
||||
outcomes[seq] = await run_tool_call(seq, tool_call)
|
||||
for seq, task in concurrent.items():
|
||||
outcomes[seq] = await task
|
||||
|
||||
tool_results: list[dict[str, Any]] = []
|
||||
for result_entry, call_entry in outcomes:
|
||||
for seq, _ in calls:
|
||||
result_entry, call_entry = outcomes[seq]
|
||||
tool_results.append(result_entry)
|
||||
if call_entry is not None:
|
||||
all_tool_calls.append(call_entry)
|
||||
|
|
|
|||
|
|
@ -2604,6 +2604,23 @@ _TOOL_HANDLERS: dict[str, Callable[[ToolContext, dict[str, Any]], Any]] = {
|
|||
"get_reasoning_chain": _handle_get_reasoning_chain,
|
||||
}
|
||||
|
||||
# Tools that change state, and so must not be reordered relative to one another
|
||||
# within a single assistant turn. Each of these routes to a handler that takes
|
||||
# `ctx.db_lock` (`_handle_create_observations_impl`, `_handle_update_peer_card`,
|
||||
# `_handle_delete_observations`); the lock stops them interleaving, but it does
|
||||
# not decide who gets it first, so the tool loop keeps them in the order the
|
||||
# model asked for. `extract_preferences` and `finish_consolidation` are not
|
||||
# here: both only return text telling the model what to call next.
|
||||
MUTATING_TOOL_NAMES: frozenset[str] = frozenset(
|
||||
{
|
||||
"create_observations",
|
||||
"create_observations_deductive",
|
||||
"create_observations_inductive",
|
||||
"update_peer_card",
|
||||
"delete_observations",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def create_tool_executor(
|
||||
workspace_name: str,
|
||||
|
|
|
|||
|
|
@ -75,8 +75,13 @@ class _Tracker:
|
|||
return "ok"
|
||||
|
||||
|
||||
async def _run(tracker: _Tracker, n_calls: int = 2) -> HonchoLLMCallResponse[Any]:
|
||||
names = ["search_memory", "search_messages"]
|
||||
async def _run(
|
||||
tracker: _Tracker,
|
||||
n_calls: int = 2,
|
||||
mutating_tools: frozenset[str] | None = frozenset(),
|
||||
names: list[str] | None = None,
|
||||
) -> HonchoLLMCallResponse[Any]:
|
||||
names = names or ["search_memory", "search_messages"]
|
||||
calls = [
|
||||
{"name": names[i % len(names)], "input": {"query": str(i)}, "id": f"t{i}"}
|
||||
for i in range(n_calls)
|
||||
|
|
@ -113,6 +118,7 @@ async def _run(tracker: _Tracker, n_calls: int = 2) -> HonchoLLMCallResponse[Any
|
|||
tool_choice="auto",
|
||||
tool_executor=tracker.execute,
|
||||
max_tool_iterations=5,
|
||||
mutating_tools=mutating_tools,
|
||||
response_model=None,
|
||||
json_mode=False,
|
||||
temperature=None,
|
||||
|
|
@ -171,3 +177,43 @@ async def test_fan_out_is_capped():
|
|||
f"fan-out reached {tracker.peak}, above the {MAX_CONCURRENT_TOOL_CALLS} cap"
|
||||
)
|
||||
assert tracker.peak > 1, "cap must not serialise execution entirely"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mutating_tools_are_not_reordered():
|
||||
"""State-changing calls keep the order the model asked for.
|
||||
|
||||
`ctx.db_lock` makes the mutating handlers take turns but does not decide
|
||||
whose turn comes first, so running them concurrently could apply a turn's
|
||||
writes in an order the model did not request.
|
||||
"""
|
||||
order: list[str] = []
|
||||
|
||||
class _Recorder(_Tracker):
|
||||
async def execute(self, name: str, _input: dict[str, Any]) -> str:
|
||||
self.in_flight += 1
|
||||
self.peak = max(self.peak, self.in_flight)
|
||||
# Later calls sleep less, so anything concurrent finishes reversed.
|
||||
await asyncio.sleep(TOOL_DELAY_SECONDS / (len(order) + 1))
|
||||
order.append(name)
|
||||
self.in_flight -= 1
|
||||
return "ok"
|
||||
|
||||
await _run(
|
||||
_Recorder(),
|
||||
n_calls=3,
|
||||
mutating_tools=frozenset({"update_peer_card"}),
|
||||
names=["update_peer_card"],
|
||||
)
|
||||
|
||||
assert order == ["update_peer_card"] * 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_default_is_fully_sequential():
|
||||
"""A caller that does not opt in keeps the old one-at-a-time behaviour."""
|
||||
tracker = _Tracker()
|
||||
|
||||
await _run(tracker, n_calls=4, mutating_tools=None)
|
||||
|
||||
assert tracker.peak == 1
|
||||
|
|
|
|||
Loading…
Reference in New Issue