mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-27 21:05:34 +00:00
Merge pull request #418 from open-jarvis/fix/managed-agent-streaming-parity
fix(server): managed-agent streaming parity — tool_calls replay, sampler params, tool DI (#382, #386, #395)
This commit is contained in:
Jon Saad-Falcon
@@ -398,6 +398,139 @@ def _resolve_tool_specs(
|
||||
return resolved
|
||||
|
||||
|
||||
# Per-agent sampler params forwarded to the engine when present in config.
|
||||
# The OpenAI-compat engine passes **kwargs straight through to the upstream
|
||||
# payload, so these reach local servers (vLLM / mlx_lm / etc.) that support
|
||||
# them. Only forwarded when explicitly set, so default agents send nothing
|
||||
# extra and engines that don't support a key never receive it. (#386)
|
||||
_SAMPLER_PARAM_KEYS = (
|
||||
"top_p",
|
||||
"top_k",
|
||||
"min_p",
|
||||
"repetition_penalty",
|
||||
"frequency_penalty",
|
||||
"presence_penalty",
|
||||
)
|
||||
|
||||
|
||||
def _sampler_kwargs(config: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Extract per-agent sampler params from a managed agent's config (#386)."""
|
||||
out: Dict[str, Any] = {}
|
||||
for key in _SAMPLER_PARAM_KEYS:
|
||||
val = config.get(key)
|
||||
if val is not None:
|
||||
out[key] = val
|
||||
return out
|
||||
|
||||
|
||||
def _replay_history_messages(
|
||||
history: List[Dict[str, Any]],
|
||||
exclude_id: str,
|
||||
) -> List[Any]:
|
||||
"""Rebuild prior-turn LLM messages from stored managed-agent history.
|
||||
|
||||
When a stored assistant turn recorded ``tool_calls``, replay them as an
|
||||
assistant tool-use message followed by the corresponding tool-result
|
||||
messages — so the model sees its own prior tool pattern instead of
|
||||
regressing to fabricated tool output on later turns. Without this, only
|
||||
the assistant's text is replayed and the tool-use signal is lost (#382).
|
||||
"""
|
||||
from openjarvis.core.types import Message, Role, ToolCall
|
||||
|
||||
messages: List[Any] = []
|
||||
for m in reversed(history):
|
||||
if m.get("id") == exclude_id:
|
||||
continue
|
||||
direction = m.get("direction")
|
||||
if direction == "user_to_agent":
|
||||
messages.append(Message(role=Role.USER, content=m.get("content") or ""))
|
||||
elif direction == "agent_to_user":
|
||||
stored = m.get("tool_calls")
|
||||
if stored:
|
||||
calls = []
|
||||
results = []
|
||||
for i, tc in enumerate(stored):
|
||||
call_id = f"hist-{m.get('id', '')}-{i}"
|
||||
calls.append(
|
||||
ToolCall(
|
||||
id=call_id,
|
||||
name=tc.get("tool", ""),
|
||||
arguments=tc.get("arguments") or "",
|
||||
)
|
||||
)
|
||||
results.append(
|
||||
Message(
|
||||
role=Role.TOOL,
|
||||
content=str(tc.get("result", "")),
|
||||
tool_call_id=call_id,
|
||||
name=tc.get("tool", ""),
|
||||
)
|
||||
)
|
||||
messages.append(
|
||||
Message(
|
||||
role=Role.ASSISTANT,
|
||||
content=m.get("content") or None,
|
||||
tool_calls=calls,
|
||||
)
|
||||
)
|
||||
messages.extend(results)
|
||||
else:
|
||||
messages.append(
|
||||
Message(role=Role.ASSISTANT, content=m.get("content") or "")
|
||||
)
|
||||
return messages
|
||||
|
||||
|
||||
def _instantiate_managed_tool(
|
||||
tool_cls: Any,
|
||||
name: str,
|
||||
*,
|
||||
engine: Any,
|
||||
model: str,
|
||||
app_state: Any,
|
||||
) -> Any:
|
||||
"""Instantiate a tool with the same dependency injection as the canonical
|
||||
``cli/ask.py`` path, so ``memory_*`` / ``channel_*`` / ``llm`` tools work
|
||||
instead of silently failing with "No backend configured" (#395).
|
||||
"""
|
||||
try:
|
||||
from openjarvis.cli.ask import (
|
||||
_CHANNEL_TOOLS,
|
||||
_MEMORY_TOOLS,
|
||||
_get_memory_backend,
|
||||
)
|
||||
except Exception: # pragma: no cover - cli import should always succeed
|
||||
_MEMORY_TOOLS = frozenset()
|
||||
_CHANNEL_TOOLS = frozenset()
|
||||
_get_memory_backend = None
|
||||
|
||||
if name in _MEMORY_TOOLS:
|
||||
backend = getattr(app_state, "memory_backend", None) if app_state else None
|
||||
if backend is None and app_state is not None and _get_memory_backend:
|
||||
cfg = getattr(app_state, "config", None)
|
||||
if cfg is not None:
|
||||
backend = _get_memory_backend(cfg)
|
||||
if backend is None:
|
||||
logger.warning(
|
||||
"Memory tool %r instantiated without a backend — calls will "
|
||||
"return no results.",
|
||||
name,
|
||||
)
|
||||
return tool_cls(backend=backend)
|
||||
if name in _CHANNEL_TOOLS:
|
||||
channel = getattr(app_state, "channel_bridge", None) if app_state else None
|
||||
if channel is None:
|
||||
logger.warning(
|
||||
"Channel tool %r instantiated without a channel — calls will "
|
||||
"fail with 'No channel backend configured'.",
|
||||
name,
|
||||
)
|
||||
return tool_cls(channel=channel)
|
||||
if name == "llm":
|
||||
return tool_cls(engine=engine, model=model)
|
||||
return tool_cls()
|
||||
|
||||
|
||||
def _build_deep_research_tools(
|
||||
engine: Any,
|
||||
model: str,
|
||||
@@ -657,15 +790,11 @@ async def _stream_managed_agent(
|
||||
if app_state is not None and dr_tools:
|
||||
app_state._dr_tools = dr_tools
|
||||
|
||||
# Load prior conversation context (DESC order, reverse for chronological)
|
||||
# Load prior conversation context (DESC order, reverse for chronological).
|
||||
# Replaying recorded tool_calls (assistant tool-use + tool results) keeps
|
||||
# multi-turn tool behaviour from regressing to fabricated output (#382).
|
||||
history = manager.list_messages(agent_id, limit=50)
|
||||
for m in reversed(history):
|
||||
if m["id"] == message_id:
|
||||
continue
|
||||
if m["direction"] == "user_to_agent":
|
||||
llm_messages.append(Message(role=Role.USER, content=m["content"]))
|
||||
elif m["direction"] == "agent_to_user":
|
||||
llm_messages.append(Message(role=Role.ASSISTANT, content=m["content"]))
|
||||
llm_messages.extend(_replay_history_messages(history, message_id))
|
||||
|
||||
# Append the current user message
|
||||
llm_messages.append(Message(role=Role.USER, content=user_content))
|
||||
@@ -960,6 +1089,10 @@ async def _stream_managed_agent(
|
||||
if resolved_tools:
|
||||
stream_kwargs["tools"] = resolved_tools
|
||||
|
||||
# Forward any per-agent sampler params (repetition_penalty, top_p, …) so
|
||||
# locally-hosted models can be tuned per agent (#386).
|
||||
stream_kwargs.update(_sampler_kwargs(config))
|
||||
|
||||
# Discover MCP tools and merge into stream_kwargs
|
||||
mcp_adapters: Dict[str, Any] = {}
|
||||
if app_state is not None:
|
||||
@@ -1175,7 +1308,17 @@ async def _stream_managed_agent(
|
||||
|
||||
tool_cls = ToolRegistry.get(tool_name)
|
||||
if tool_cls is not None:
|
||||
tool_instance = tool_cls()
|
||||
# Inject backend / channel / engine the same
|
||||
# way cli/ask.py does, else memory_* / channel_*
|
||||
# / llm tools fail with "No backend configured"
|
||||
# on every call (#395).
|
||||
tool_instance = _instantiate_managed_tool(
|
||||
tool_cls,
|
||||
tool_name,
|
||||
engine=engine,
|
||||
model=model,
|
||||
app_state=app_state,
|
||||
)
|
||||
# Tools the user explicitly added to this
|
||||
# agent's toolkit are considered pre-approved —
|
||||
# selecting them in the wizard is the
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
"""Managed-agent streaming parity tests (#382, #386, #395).
|
||||
|
||||
These exercise the pure helpers extracted from ``_stream_managed_agent`` so the
|
||||
streaming path's history replay, sampler forwarding, and tool dependency
|
||||
injection can be verified without a live engine.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
pytest.importorskip("fastapi") # agent_manager_routes imports FastAPI at module load
|
||||
|
||||
from openjarvis.core.types import Role # noqa: E402
|
||||
from openjarvis.server.agent_manager_routes import ( # noqa: E402
|
||||
_instantiate_managed_tool,
|
||||
_replay_history_messages,
|
||||
_sampler_kwargs,
|
||||
)
|
||||
|
||||
|
||||
class TestReplayHistoryToolCalls:
|
||||
"""#382 — stored tool_calls must be replayed, not dropped."""
|
||||
|
||||
def test_assistant_tool_calls_are_replayed_with_results(self):
|
||||
history = [ # DESC order (newest first), as list_messages returns
|
||||
{"id": "m2", "direction": "agent_to_user", "content": "", "tool_calls": [
|
||||
{"tool": "shell_exec", "arguments": '{"command":"pwd"}',
|
||||
"result": "/home/u", "success": True, "latency": 1.0},
|
||||
]},
|
||||
{"id": "m1", "direction": "user_to_agent", "content": "run pwd",
|
||||
"tool_calls": None},
|
||||
]
|
||||
msgs = _replay_history_messages(history, exclude_id="current")
|
||||
# Chronological: user, assistant(tool_calls), tool(result)
|
||||
assert [m.role for m in msgs] == [Role.USER, Role.ASSISTANT, Role.TOOL]
|
||||
assistant = msgs[1]
|
||||
assert assistant.tool_calls is not None
|
||||
assert assistant.tool_calls[0].name == "shell_exec"
|
||||
tool_msg = msgs[2]
|
||||
assert tool_msg.content == "/home/u"
|
||||
# The tool result must reference the assistant's tool_call id.
|
||||
assert tool_msg.tool_call_id == assistant.tool_calls[0].id
|
||||
|
||||
def test_plain_assistant_without_tool_calls(self):
|
||||
history = [
|
||||
{"id": "m1", "direction": "agent_to_user", "content": "hi",
|
||||
"tool_calls": None},
|
||||
]
|
||||
msgs = _replay_history_messages(history, exclude_id="current")
|
||||
assert len(msgs) == 1
|
||||
assert msgs[0].role == Role.ASSISTANT
|
||||
assert msgs[0].tool_calls is None
|
||||
|
||||
def test_excludes_current_message(self):
|
||||
history = [
|
||||
{"id": "cur", "direction": "user_to_agent", "content": "now",
|
||||
"tool_calls": None},
|
||||
{"id": "old", "direction": "user_to_agent", "content": "before",
|
||||
"tool_calls": None},
|
||||
]
|
||||
msgs = _replay_history_messages(history, exclude_id="cur")
|
||||
assert [m.content for m in msgs] == ["before"]
|
||||
|
||||
|
||||
class TestSamplerKwargs:
|
||||
"""#386 — sampler params forwarded only when set."""
|
||||
|
||||
def test_reads_present_keys(self):
|
||||
cfg = {"temperature": 0.7, "repetition_penalty": 1.1, "top_p": 0.9,
|
||||
"top_k": 40, "min_p": 0.05, "frequency_penalty": 0.2,
|
||||
"presence_penalty": 0.1}
|
||||
out = _sampler_kwargs(cfg)
|
||||
assert out == {"repetition_penalty": 1.1, "top_p": 0.9, "top_k": 40,
|
||||
"min_p": 0.05, "frequency_penalty": 0.2,
|
||||
"presence_penalty": 0.1}
|
||||
# temperature/max_tokens are handled separately, not here.
|
||||
assert "temperature" not in out
|
||||
|
||||
def test_omits_absent_keys(self):
|
||||
assert _sampler_kwargs({"temperature": 0.7}) == {}
|
||||
|
||||
def test_none_values_skipped(self):
|
||||
assert _sampler_kwargs({"top_p": None, "repetition_penalty": 1.05}) == {
|
||||
"repetition_penalty": 1.05
|
||||
}
|
||||
|
||||
|
||||
class _FakeTool:
|
||||
def __init__(self, **kwargs):
|
||||
self.kwargs = kwargs
|
||||
|
||||
|
||||
class TestInstantiateManagedTool:
|
||||
"""#395 — tools get their dependencies injected."""
|
||||
|
||||
def test_memory_tool_gets_backend(self):
|
||||
backend = object()
|
||||
app_state = SimpleNamespace(memory_backend=backend, config=None,
|
||||
channel_bridge=None)
|
||||
tool = _instantiate_managed_tool(_FakeTool, "memory_store",
|
||||
engine=object(), model="m",
|
||||
app_state=app_state)
|
||||
assert tool.kwargs == {"backend": backend}
|
||||
|
||||
def test_llm_tool_gets_engine_and_model(self):
|
||||
engine = object()
|
||||
app_state = SimpleNamespace(memory_backend=None, config=None,
|
||||
channel_bridge=None)
|
||||
tool = _instantiate_managed_tool(_FakeTool, "llm", engine=engine,
|
||||
model="qwen", app_state=app_state)
|
||||
assert tool.kwargs == {"engine": engine, "model": "qwen"}
|
||||
|
||||
def test_channel_tool_gets_channel(self):
|
||||
bridge = object()
|
||||
app_state = SimpleNamespace(memory_backend=None, config=None,
|
||||
channel_bridge=bridge)
|
||||
tool = _instantiate_managed_tool(_FakeTool, "channel_send",
|
||||
engine=object(), model="m",
|
||||
app_state=app_state)
|
||||
assert tool.kwargs == {"channel": bridge}
|
||||
|
||||
def test_plain_tool_gets_no_injection(self):
|
||||
app_state = SimpleNamespace(memory_backend=None, config=None,
|
||||
channel_bridge=None)
|
||||
tool = _instantiate_managed_tool(_FakeTool, "calculator",
|
||||
engine=object(), model="m",
|
||||
app_state=app_state)
|
||||
assert tool.kwargs == {}
|
||||
Reference in New Issue
Block a user