xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-11-05 07:12:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(localagent): address review blockers for progressive disclosure (LANG-233)

Browse Source 
- Replace SDK resource imports in tests with SimpleNamespace/dicts for 0.7.19 collection compatibility
- Add tool call started/completed tracing via _TracedRunAPI results queue for workspace tools
- Update simulated e2e tests and operation docs fixture to match real Core catalog metadata
- Add hash fallback for operation IDs longer than 64 characters
- Guard against reserved argument keywords (operation_id, _timeout) in workspace_call
- Remove unused build_workspace_tools_system_message helper
- Declare workspace-direct-threshold in default.yaml and clean config aliases

Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
RockChinQ
2026-11-01 08:15:18 +00:00
co-authored by multica-agent
parent 2b3770ed26
commit 1400bb8cfb
6 changed files with 321 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/Runner/LocalAgent/components/runner/default.yaml
+14
View File
@@ -247,6 +247,20 @@ spec:
field: advanced-settings
operator: eq
value: true
- name: workspace-direct-threshold
label:
en_US: Workspace Operations Direct Mode Threshold
zh_Hans: 工作区操作直接模式阈值
description:
en_US: Maximum number of authorized workspace operations to expose directly as individual tools before switching to progressive search/describe/call tools.
zh_Hans: 直接作为独立工具暴露给模型的工作区操作数量上限。超过此阈值时切换为搜索/描述/调用的渐进式披露模式。
type: integer
required: false
default: 8
show_if:
field: advanced-settings
operator: eq
value: true
- name: context-history-fetch-limit
label:
en_US: History Fetch Limit
plugins/Runner/LocalAgent/pkg/agent_core/langbot.py
+74 -3
View File
@@ -6,9 +6,11 @@ import json
import logging
import re
import typing
import uuid
from langbot_plugin.api.entities.builtin.provider.message import ContentElement, Message, MessageChunk
from langbot_plugin.api.entities.builtin.resource.tool import LLMTool
from langbot_plugin.api.entities.builtin.runner.result import RunnerResult
from langbot_plugin.api.proxies.runner import PermissionDeniedError, RunnerAPIProxy
from pkg.config import DEFAULT_MAX_TOOL_RESULT_CHARS
@@ -246,7 +248,27 @@ class LangBotToolExecutor:
if prepared.error is not None:
return self.finalize(prepared, result=None, error=prepared.error)
results_queue = getattr(self.api, "_results", None)
raw_run_id = getattr(self.api, "_run_id", None)
run_id = raw_run_id if isinstance(raw_run_id, str) and raw_run_id else "local-run"
call_id = prepared.request.id or str(uuid.uuid4())
is_traced_queue = (
results_queue is not None
and hasattr(results_queue, "put")
and hasattr(results_queue, "get")
and type(results_queue).__name__ not in ("Mock", "AsyncMock", "MagicMock")
)
if self.workspace_manager is not None and self.workspace_manager.is_workspace_tool(prepared.request.name):
if is_traced_queue:
await results_queue.put(
RunnerResult.tool_call_started(
run_id,
call_id,
prepared.request.name,
prepared.parameters,
)
)
try:
result = await self.workspace_manager.execute(
api=self.api,
@@ -254,16 +276,65 @@ class LangBotToolExecutor:
parameters=prepared.parameters,
max_result_chars=self.max_result_chars,
)
if is_traced_queue:
await results_queue.put(
RunnerResult.tool_call_completed(
run_id,
call_id,
prepared.request.name,
result=result if isinstance(result, dict) else {"value": result},
)
)
return self.finalize(prepared, result=result, error=None)
except Exception as e:
if is_deadline_exceeded_error(e):
raise
return self.finalize(prepared, result=None, error=f"Workspace tool execution failed: {e}")
error_msg = f"Workspace tool execution failed: {e}"
if is_traced_queue:
await results_queue.put(
RunnerResult.tool_call_completed(
run_id,
call_id,
prepared.request.name,
error=error_msg,
)
)
return self.finalize(prepared, result=None, error=error_msg)
try:
if prepared.request.name in self.local_tools:
result = await self.local_tools[prepared.request.name](**prepared.parameters)
return self.finalize(prepared, result=result, error=None)
if is_traced_queue:
await results_queue.put(
RunnerResult.tool_call_started(
run_id,
call_id,
prepared.request.name,
prepared.parameters,
)
)
try:
result = await self.local_tools[prepared.request.name](**prepared.parameters)
if is_traced_queue:
await results_queue.put(
RunnerResult.tool_call_completed(
run_id,
call_id,
prepared.request.name,
result=result if isinstance(result, dict) else {"value": result},
)
)
return self.finalize(prepared, result=result, error=None)
except Exception as exc:
if is_traced_queue:
await results_queue.put(
RunnerResult.tool_call_completed(
run_id,
call_id,
prepared.request.name,
error=str(exc),
)
)
raise
result = await self.api.call_tool(
tool_name=prepared.request.name,
parameters=prepared.parameters,
plugins/Runner/LocalAgent/pkg/config.py
-6
View File
@@ -191,12 +191,6 @@ def get_workspace_direct_threshold(config: dict[str, typing.Any] | None) -> int:
if not isinstance(config, dict):
return DEFAULT_WORKSPACE_DIRECT_THRESHOLD
val = config.get("workspace-direct-threshold")
if val is None:
val = config.get("workspace_direct_threshold")
if val is None:
val = config.get("workspace-tools-threshold")
if val is None:
val = config.get("workspace_tools_threshold")
if isinstance(val, bool) or not isinstance(val, int) or val < 0:
return DEFAULT_WORKSPACE_DIRECT_THRESHOLD
return val
plugins/Runner/LocalAgent/pkg/messages.py
-15
View File
@@ -145,21 +145,6 @@ def build_platform_tools_system_message(ctx: typing.Any) -> Message | None:
)
def build_workspace_tools_system_message(
domains: list[dict[str, typing.Any]] | None,
max_chars: int = 1500,
) -> Message | None:
"""Build a system message with progressive disclosure instructions and domain overview."""
from pkg.workspace_tools import PROGRESSIVE_GUIDANCE, format_domains_overview
overview = format_domains_overview(domains or [], max_chars=max_chars)
if overview:
content = f"{PROGRESSIVE_GUIDANCE}\n\n{overview}"
else:
content = PROGRESSIVE_GUIDANCE
return Message(role="system", content=content)
def build_user_message(
user_text: str,
input_contents: list[ContentElement] | None = None,
plugins/Runner/LocalAgent/pkg/workspace_tools.py
+32 -6
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import collections.abc
import hashlib
import json
import logging
from dataclasses import dataclass
@@ -113,10 +114,13 @@ def operation_id_to_tool_name(operation_id: str) -> str:
"""Convert operation ID to direct tool name: 'bot.list' -> 'ws_bot__list'.
Tool name starts with 'ws_', replaces '.' with '__', and length is bounded <= 64.
If length exceeds 64 chars, a stable 8-char SHA256 suffix is appended to prevent collision.
"""
tool_name = f"ws_{operation_id.replace('.', '__')}"
cleaned = operation_id.replace(".", "__")
tool_name = f"ws_{cleaned}"
if len(tool_name) > 64:
return tool_name[:64]
digest = hashlib.sha256(operation_id.encode("utf-8")).hexdigest()[:8]
tool_name = f"{tool_name[:55]}_{digest}"
return tool_name
@@ -226,6 +230,7 @@ class WorkspaceToolsManager:
domains: list[dict[str, Any]],
mode: str | None, # "direct", "progressive", or None
tools: list[LLMTool],
tool_name_to_op_id: dict[str, str] | None = None,
):
self.operations = operations
self.operations_by_id = {op.id: op for op in operations}
@@ -233,6 +238,7 @@ class WorkspaceToolsManager:
self.mode = mode
self.tools = tools
self._tool_names = {tool.name for tool in tools}
self._tool_name_to_op_id = dict(tool_name_to_op_id or {})
@classmethod
async def discover(
@@ -282,9 +288,10 @@ class WorkspaceToolsManager:
config = getattr(ctx, "config", None) if isinstance(getattr(ctx, "config", None), dict) else {}
threshold = get_workspace_direct_threshold(config)
tool_name_map: dict[str, str] = {}
if len(normalized_ops) <= threshold:
mode = "direct"
tools = cls._build_direct_tools(normalized_ops)
tools, tool_name_map = cls._build_direct_tools(normalized_ops)
else:
mode = "progressive"
tools = cls._build_progressive_tools()
@@ -294,18 +301,23 @@ class WorkspaceToolsManager:
domains=domains,
mode=mode,
tools=tools,
tool_name_to_op_id=tool_name_map,
)
@staticmethod
def _build_direct_tools(operations: list[NormalizedOperation]) -> list[LLMTool]:
def _build_direct_tools(
operations: list[NormalizedOperation],
) -> tuple[list[LLMTool], dict[str, str]]:
"""Build direct tools: ws_<op_id with . replaced by __>."""
async def _placeholder(**kwargs: Any) -> Any:
return kwargs
tools: list[LLMTool] = []
name_map: dict[str, str] = {}
for op in operations:
tool_name = operation_id_to_tool_name(op.id)
name_map[tool_name] = op.id
desc = op.description or op.summary or f"Workspace operation {op.id}"
params = (
op.input_schema
@@ -324,7 +336,7 @@ class WorkspaceToolsManager:
func=_placeholder,
)
)
return tools
return tools, name_map
@staticmethod
def _build_progressive_tools() -> list[LLMTool]:
@@ -540,7 +552,7 @@ class WorkspaceToolsManager:
else:
return {"error": f"Unknown workspace tool '{tool_name}'"}
elif self.mode == "direct":
op_id = tool_name_to_operation_id(tool_name)
op_id = self._tool_name_to_op_id.get(tool_name) or tool_name_to_operation_id(tool_name)
return await self._call_operation(
api=api,
operation_id=op_id,
@@ -567,6 +579,20 @@ class WorkspaceToolsManager:
},
}
# Guard against reserved keyword collision in SDK WorkspaceOperationsProxy.call(operation_id, _timeout, **arguments)
for reserved_key in ("operation_id", "_timeout"):
if reserved_key in arguments:
return {
"ok": False,
"error": {
"code": "invalid_argument",
"message": (
f"Argument conflict: '{reserved_key}' is a reserved parameter in workspace.call. "
"Nested parameters should be used or this parameter should be renamed."
),
},
}
workspace_proxy = getattr(api, "workspace", None)
if workspace_proxy is None or not callable(getattr(workspace_proxy, "call", None)):
return {
plugins/Runner/LocalAgent/tests/test_workspace_tools.py
+201 -30
View File
@@ -2,6 +2,9 @@
from __future__ import annotations
import asyncio
import hashlib
import types
from typing import Any
from unittest.mock import AsyncMock, Mock
@@ -12,7 +15,6 @@ from langbot_plugin.api.entities.builtin.runner.resources import (
AgentResources,
ModelResource,
ToolResource,
WorkspaceOperationResource,
)
from langbot_plugin.api.entities.builtin.runner.result import (
RunnerResultType,
@@ -37,9 +39,9 @@ def _make_op(
description: str = "",
risk: str = "read",
input_schema: dict[str, Any] | None = None,
) -> WorkspaceOperationResource:
) -> types.SimpleNamespace:
d = domain or op_id.split(".")[0]
return WorkspaceOperationResource(
return types.SimpleNamespace(
id=op_id,
domain=d,
label={"zh_Hans": d, "en_US": d.capitalize()},
@@ -51,7 +53,7 @@ def _make_op(
def _make_ctx(
workspace_ops: list[WorkspaceOperationResource] | None = None,
workspace_ops: list[Any] | None = None,
tools: list[ToolResource] | None = None,
config: dict[str, Any] | None = None,
models: list[ModelResource] | None = None,
@@ -65,8 +67,9 @@ def _make_ctx(
resources = AgentResources(
models=models,
tools=tools or [],
workspace_operations=workspace_ops if workspace_ops is not None else [],
)
if workspace_ops is not None:
setattr(resources, "workspace_operations", workspace_ops)
return make_context(
run_id="run-test-1",
input_text="test question",
@@ -400,16 +403,143 @@ async def test_tool_executor_handles_workspace_tools():
# ==============================================================================
# 11. 端到端本地运行记录：直接模式与渐进模式
# 11. 追踪与可观测性（_TracedRunAPI 队列事件）
# ==============================================================================
@pytest.mark.asyncio
async def test_tool_executor_emits_traced_run_api_events():
"""Verify that when api is wrapped by _TracedRunAPI (having _results queue and _run_id),
executing a workspace tool emits tool_call_started and tool_call_completed events to the queue.
"""
ops = [_make_op("bot.list", summary="List bots")]
ctx = _make_ctx(workspace_ops=ops)
api = Mock()
api.workspace = Mock()
api.workspace.call = AsyncMock(return_value={"ok": True, "result": ["bot-1"]})
# Simulate _TracedRunAPI wrapper probe
results_queue = asyncio.Queue()
api._results = results_queue
api._run_id = "test-run-trace-123"
mgr = await WorkspaceToolsManager.discover(api, ctx)
executor = LangBotToolExecutor(
api=api,
allowed_tools={"ws_bot__list"},
max_result_chars=20000,
workspace_manager=mgr,
)
prep = executor.prepare(
ToolCallRequest(
id="call-trace-1",
name="ws_bot__list",
arguments='{"limit": 5}',
)
)
outcome = await executor.execute(prep)
assert outcome.error is None
# Verify queue received 2 events (started and completed)
assert results_queue.qsize() == 2
started = await results_queue.get()
completed = await results_queue.get()
assert started.type == RunnerResultType.TOOL_CALL_STARTED
assert started.run_id == "test-run-trace-123"
assert started.data["tool_name"] == "ws_bot__list"
assert started.data["parameters"] == {"limit": 5}
assert completed.type == RunnerResultType.TOOL_CALL_COMPLETED
assert completed.run_id == "test-run-trace-123"
assert completed.data["tool_name"] == "ws_bot__list"
assert completed.data["result"] == {"ok": True, "result": ["bot-1"]}
# ==============================================================================
# 12. 超长操作名哈希后缀与保留参数校验
# ==============================================================================
def test_long_operation_id_hashing():
"""Verify operation ID > 64 chars uses a stable 8-char SHA256 suffix bounded <= 64."""
very_long_op = "pipeline.very_long_domain_name.subdomain_action_that_exceeds_sixty_four_characters_easily"
tool_name = operation_id_to_tool_name(very_long_op)
assert len(tool_name) <= 64
assert tool_name.startswith("ws_")
expected_digest = hashlib.sha256(very_long_op.encode("utf-8")).hexdigest()[:8]
assert tool_name.endswith(expected_digest)
@pytest.mark.asyncio
async def test_reserved_argument_conflict_check():
"""Verify that passing 'operation_id' or '_timeout' inside arguments returns invalid_argument error."""
ops = [_make_op("bot.list")]
ctx = _make_ctx(workspace_ops=ops)
api = Mock()
api.workspace = Mock()
mgr = await WorkspaceToolsManager.discover(api, ctx)
res = await mgr.execute(api, "ws_bot__list", {"operation_id": "malicious", "limit": 1})
assert res["ok"] is False
assert res["error"]["code"] == "invalid_argument"
assert "reserved parameter" in res["error"]["message"]
@pytest.mark.asyncio
async def test_sdk_0719_graceful_degradation_without_workspace_attr():
"""Verify that when running on 0.7.19 where resources has no workspace_operations
and api has no workspace attribute, discovery gracefully returns mode=None and 0 tools."""
api_0719 = Mock(spec=["call_tool", "get_allowed_tools"])
ctx_0719 = make_context(
run_id="run-0719",
input_text="hello",
resources=AgentResources(
models=[ModelResource(model_id="primary", model_name="Primary")],
tools=[],
),
)
if hasattr(ctx_0719.resources, "workspace_operations"):
delattr(ctx_0719.resources, "workspace_operations")
mgr = await WorkspaceToolsManager.discover(api_0719, ctx_0719)
assert mgr.mode is None
assert len(mgr.tools) == 0
assert mgr.build_system_message() is None
# ==============================================================================
# 13. 端到端本地运行记录：直接模式与渐进模式（真实 Core Catalog 元数据）
# ==============================================================================
@pytest.mark.asyncio
async def test_e2e_direct_mode_agent_run():
"""Simulated local agent run in direct mode."""
"""Simulated local agent run in direct mode with real Core catalog metadata."""
ops = [
_make_op("bot.list", summary="List bots"),
_make_op("stats.basic", summary="Basic usage stats"),
_make_op(
"bot.list",
summary="List all configured chat bots in the workspace.",
description=(
"Retrieve all configured chat bots in the workspace with basic settings, "
"adapter types, and pipeline bindings. Set include_secret=True to include credentials."
),
risk="read",
input_schema={
"type": "object",
"properties": {
"limit": {"type": "integer"},
"include_secret": {"type": "boolean"},
},
},
),
_make_op(
"stats.basic",
summary="Get workspace overview counters for bots, models, and pipelines.",
description="Retrieve aggregate workspace statistics including total bot count, active pipelines, configured models, and knowledge bases.",
risk="read",
),
]
ctx = _make_ctx(workspace_ops=ops)
@@ -419,9 +549,10 @@ async def test_e2e_direct_mode_agent_run():
)
api.get_prompt = AsyncMock(return_value=[{"role": "system", "content": "You are a test assistant."}])
api.workspace = Mock()
api.workspace.call = AsyncMock(return_value={"ok": True, "result": {"bots": [{"id": "b1", "name": "HelperBot"}]}})
api.workspace.call = AsyncMock(
return_value={"ok": True, "result": [{"id": "bot-1", "name": "SupportBot"}, {"id": "bot-2", "name": "DocsBot"}]}
)
# Model mock: turn 1 calls ws_bot__list; turn 2 responds with text
turn_1_msg = Message(
role="assistant",
content="I will check the bots.",
@@ -429,13 +560,13 @@ async def test_e2e_direct_mode_agent_run():
{
"id": "call-bot-list-1",
"type": "function",
"function": {"name": "ws_bot__list", "arguments": '{"limit": 5}'},
"function": {"name": "ws_bot__list", "arguments": '{"limit": 10}'},
}
],
)
turn_2_msg = Message(
role="assistant",
content="There is 1 bot available: HelperBot (id: b1).",
content="There are 2 bots in this workspace:\n1. SupportBot (ID: bot-1)\n2. DocsBot (ID: bot-2)",
)
call_count = 0
@@ -464,20 +595,58 @@ async def test_e2e_direct_mode_agent_run():
assert len(tool_started) == 1
assert tool_started[0].data["tool_name"] == "ws_bot__list"
assert tool_started[0].data["parameters"] == {"limit": 5}
assert tool_started[0].data["parameters"] == {"limit": 10}
assert len(tool_completed) == 1
assert tool_completed[0].data["tool_name"] == "ws_bot__list"
assert len(msg_completed) == 1
assert "HelperBot" in msg_completed[0].data["message"]["content"]
assert "SupportBot" in msg_completed[0].data["message"]["content"]
assert len(run_completed) == 1
@pytest.mark.asyncio
async def test_e2e_progressive_mode_agent_run():
"""Simulated local agent run in progressive mode with search -> describe -> call."""
ops = [_make_op(f"op_{i}.action_{i}", summary=f"Action {i} for testing") for i in range(15)]
"""Simulated local agent run in progressive mode with real Core catalog metadata."""
ops = [
_make_op(
"pipeline.copy",
domain="pipeline",
summary="Duplicate an existing pipeline to create a new copy.",
description="Clone an existing pipeline definition under a new identifier. Prerequisites: Call pipeline.list to obtain source pipeline_uuid. Side effect: Creates duplicate pipeline.",
risk="write",
input_schema={
"type": "object",
"properties": {
"pipeline_uuid": {"type": "string"},
"new_name": {"type": "string"},
},
"required": ["pipeline_uuid"],
},
),
_make_op(
"pipeline.delete",
domain="pipeline",
summary="Permanently delete a pipeline from the workspace.",
description="Delete a processing pipeline and disconnect it from associated bots. Prerequisites: Call pipeline.list to obtain pipeline_uuid. Side effect: Irreversible removal of pipeline.",
risk="dangerous",
input_schema={
"type": "object",
"properties": {
"pipeline_uuid": {"type": "string"},
},
"required": ["pipeline_uuid"],
},
),
_make_op("pipeline.list", summary="List all processing pipelines in the workspace.", risk="read"),
_make_op("pipeline.get", summary="Get pipeline details and flow configuration.", risk="read"),
_make_op("pipeline.create", summary="Create a new processing pipeline.", risk="write"),
_make_op("bot.list", summary="List all configured chat bots.", risk="read"),
_make_op("bot.create", summary="Create a new bot.", risk="write"),
_make_op("bot.delete", summary="Delete a bot from the workspace.", risk="dangerous"),
_make_op("knowledge.list", summary="List all knowledge bases.", risk="read"),
_make_op("model.llm.list", summary="List configured language models.", risk="read"),
]
ctx = _make_ctx(workspace_ops=ops)
api = FakeRunnerAPIProxy(
@@ -486,15 +655,17 @@ async def test_e2e_progressive_mode_agent_run():
)
api.get_prompt = AsyncMock(return_value=[{"role": "system", "content": "You are a test assistant."}])
api.workspace = Mock()
api.workspace.call = AsyncMock(
return_value={"ok": True, "result": {"success": True, "details": "Operation executed"}}
)
# 3-step progressive execution:
# Turn 1: workspace_search(query="action_1")
# Turn 2: workspace_describe(operation_id="op_1.action_1")
# Turn 3: workspace_call(operation_id="op_1.action_1", arguments={"param": 42})
# Turn 4: Final response
async def mock_call(operation_id, **arguments):
if operation_id == "pipeline.copy":
return {"ok": True, "result": {"pipeline_uuid": "pipe-copy-uuid", "name": "pipe-prod-copy"}}
elif operation_id == "pipeline.delete":
return {"ok": True, "result": {"deleted": True}}
return {"ok": True}
api.workspace.call = AsyncMock(side_effect=mock_call)
# 4-turn progressive execution:
turn_1 = Message(
role="assistant",
content="Searching operations...",
@@ -502,7 +673,7 @@ async def test_e2e_progressive_mode_agent_run():
{
"id": "c-1",
"type": "function",
"function": {"name": "workspace_search", "arguments": '{"query": "action_1"}'},
"function": {"name": "workspace_search", "arguments": '{"query": "copy", "domain": "pipeline"}'},
}
],
)
@@ -513,7 +684,7 @@ async def test_e2e_progressive_mode_agent_run():
{
"id": "c-2",
"type": "function",
"function": {"name": "workspace_describe", "arguments": '{"operation_id": "op_1.action_1"}'},
"function": {"name": "workspace_describe", "arguments": '{"operation_id": "pipeline.copy"}'},
}
],
)
@@ -526,14 +697,14 @@ async def test_e2e_progressive_mode_agent_run():
"type": "function",
"function": {
"name": "workspace_call",
"arguments": '{"operation_id": "op_1.action_1", "arguments": {"param": 42}}',
"arguments": '{"operation_id": "pipeline.copy", "arguments": {"pipeline_uuid": "pipe-prod-uuid", "new_name": "pipe-prod-copy"}}',
},
}
],
)
turn_4 = Message(
role="assistant",
content="Operation op_1.action_1 successfully completed.",
content="Pipeline copy created successfully: pipe-prod-copy (UUID: pipe-copy-uuid).",
)
turns = [turn_1, turn_2, turn_3, turn_4]
@@ -564,5 +735,5 @@ async def test_e2e_progressive_mode_agent_run():
assert called_tool_names == ["workspace_search", "workspace_describe", "workspace_call"]
assert len(tool_completed) == 3
assert len(msg_completed) == 1
assert "op_1.action_1 successfully completed" in msg_completed[0].data["message"]["content"]
assert "pipe-copy-uuid" in msg_completed[0].data["message"]["content"]
assert len(run_completed) == 1