core/agent+cli: reasoning deltas, /stream and /verbose

Decode and stream reasoning_content as ReasoningDeltaEvent; keep it on
assistant messages. CLI shows dim reasoning, toggles stream/verbose, and
prints full tool results when verbose.
This commit is contained in:
2026-07-15 12:11:19 +08:00
parent e566da596b
commit 04d421a4b5
14 changed files with 272 additions and 18 deletions
+2 -2
View File
@@ -57,7 +57,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `
- **`run_chat_loop`**: multi-round tool loop; default **streaming** text deltas + stream tool-call merge; parallel tools; tool-result char budget; soft context compact on request (**API-calibrated** after first usage when available); cooperative cancel points; optional `on_limit`.
- **`ChatAgent`**: optional `MemoryStore` (user message persisted immediately; assistant/tools on success); `stream`; system prompt; `retry()` when history ends with a user message (failed/cancelled turn or orphan after resume).
- **`/compact`**: soft-compact tool dumps → model summary (no tools) → **new** session seeded with summary message.
- Events: text_delta, assistant_message, tool_call/result, max_rounds, **error** (`retryable`/`source`), **cancelled** (`reason`), **usage** (`TokenUsage`).
- Events: text_delta, **reasoning_delta**, assistant_message, tool_call/result, max_rounds, **error** (`retryable`/`source`), **cancelled** (`reason`), **usage** (`TokenUsage`).
- Usage: API `usage` from completions (stream with `include_usage`); **char≈token fallback** (~4 chars/token) when omitted; **context size** = last request ``prompt_tokens`` (API preferred); `last_turn_usage` / `session_usage` are **billed sums** (tool rounds re-send history); CLI end-of-turn + `/status`.
- Config ``[agent]``: `system_prompt`, `max_tool_result_chars`, `parallel_tools`, `confirm_destructive`, `path_denylist`, `max_context_tokens` (default 200k est. tokens).
@@ -84,7 +84,7 @@ Shared interactive I/O: `ask` / `choose` / `form` / `confirm` with pluggable bac
Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
- **`plyngent chat`**: provider/model selection (flags or interactive), SQLite sessions via config `[database]` (file DB under user data if unset/`:memory:`), sessions bound to workspace dir; resumes **most recently updated** session for cwd/`--workspace` by default (`--new` / `--session`).
- Slash: `/history`, `/sessions` (newest first), `/resume [id]`, `/compact`, `/status` (incl. context char estimate), `/rounds`, `/retry`, …
- Slash: `/history`, `/sessions` (newest first), `/resume [id]`, `/compact`, `/status` (incl. context char estimate), `/rounds`, `/stream`, `/verbose`, `/retry`, …
- Explicit `/resume` or `--session` from another workspace prompts: **keep** session path, **update** binding to current, or **abort**.
- Failed/cancelled turns: user message kept in DB; partial assistant/tool rolled back; Ctrl+C cancels in-flight turn; **TTY confirms** off-loop; auto-retry 10s/20s/30s then `/retry` (no duplicate user message).
- **`plyngent providers`**: list config providers.
+1
View File
@@ -7,6 +7,7 @@ from .events import AssistantMessageEvent as AssistantMessageEvent
from .events import CancelledEvent as CancelledEvent
from .events import ErrorEvent as ErrorEvent
from .events import MaxRoundsEvent as MaxRoundsEvent
from .events import ReasoningDeltaEvent as ReasoningDeltaEvent
from .events import TextDeltaEvent as TextDeltaEvent
from .events import ToolCallEvent as ToolCallEvent
from .events import ToolResultEvent as ToolResultEvent
+5
View File
@@ -13,6 +13,10 @@ class TextDeltaEvent(Struct, tag_field="type", tag="text_delta"):
content: str
class ReasoningDeltaEvent(Struct, tag_field="type", tag="reasoning_delta"):
content: str
class AssistantMessageEvent(Struct, tag_field="type", tag="assistant_message"):
message: AssistantChatMessage
@@ -51,6 +55,7 @@ class UsageEvent(Struct, tag_field="type", tag="usage"):
type AgentEvent = (
TextDeltaEvent
| ReasoningDeltaEvent
| AssistantMessageEvent
| ToolCallEvent
| ToolResultEvent
+10
View File
@@ -31,6 +31,7 @@ from .events import (
AssistantMessageEvent,
ErrorEvent,
MaxRoundsEvent,
ReasoningDeltaEvent,
TextDeltaEvent,
ToolCallEvent,
ToolResultEvent,
@@ -122,6 +123,9 @@ async def _non_stream_round(
msg = "chat completion response contained no choices"
raise RuntimeError(msg)
assistant = response.choices[0].message
reasoning = assistant.reasoning_content
if isinstance(reasoning, str) and reasoning:
yield ReasoningDeltaEvent(content=reasoning)
if isinstance(assistant.content, str) and assistant.content:
yield TextDeltaEvent(content=assistant.content)
yield AssistantMessageEvent(message=assistant)
@@ -147,6 +151,7 @@ async def _stream_round(
)
stream = await client.chat_completions(stream_param, stream=True)
content_parts: list[str] = []
reasoning_parts: list[str] = []
tool_deltas: list[StreamToolCallDelta] = []
last_api_usage: object = UNSET
@@ -158,6 +163,9 @@ async def _stream_round(
continue
choice = chunk.choices[0]
delta = choice.delta
if isinstance(delta.reasoning_content, str) and delta.reasoning_content:
reasoning_parts.append(delta.reasoning_content)
yield ReasoningDeltaEvent(content=delta.reasoning_content)
if isinstance(delta.content, str) and delta.content:
content_parts.append(delta.content)
yield TextDeltaEvent(content=delta.content)
@@ -165,6 +173,7 @@ async def _stream_round(
tool_deltas.extend(delta.tool_calls)
full_content = "".join(content_parts)
full_reasoning = "".join(reasoning_parts)
tool_calls: list[AnyAssistantToolCall] | Unset = UNSET
if tool_deltas:
calls = merge_stream_tool_calls(tool_deltas)
@@ -174,6 +183,7 @@ async def _stream_round(
assistant = AssistantChatMessage(
content=full_content or None,
tool_calls=tool_calls,
reasoning_content=full_reasoning or UNSET,
)
yield AssistantMessageEvent(message=assistant)
yield UsageEvent(
+39 -7
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from contextvars import ContextVar
from typing import TYPE_CHECKING
import click
@@ -8,6 +9,7 @@ from plyngent.agent import (
CancelledEvent,
ErrorEvent,
MaxRoundsEvent,
ReasoningDeltaEvent,
TextDeltaEvent,
ToolCallEvent,
ToolResultEvent,
@@ -23,6 +25,18 @@ if TYPE_CHECKING:
_TOOL_RESULT_PREVIEW = 120
_TOOL_ARGS_PREVIEW = 80
# Process/session display flag for tool result dumps (set from ReplState / slash).
_verbose_tool_results: ContextVar[bool] = ContextVar("verbose_tool_results", default=False)
def set_verbose_tool_results(enabled: bool) -> None: # noqa: FBT001
"""Set whether tool results print in full (True) or as a short preview."""
_ = _verbose_tool_results.set(enabled)
def get_verbose_tool_results() -> bool:
return _verbose_tool_results.get()
def _preview(text: str, limit: int) -> str:
if len(text) <= limit:
@@ -41,11 +55,25 @@ def _echo_stream(text: str) -> None:
pass
async def render_events(events: AsyncIterator[AgentEvent]) -> None: # noqa: C901, PLR0912
async def render_events( # noqa: C901, PLR0912, PLR0915
events: AsyncIterator[AgentEvent],
*,
verbose: bool | None = None,
) -> None:
"""Print agent events to the terminal (text deltas stream as they arrive)."""
show_full = get_verbose_tool_results() if verbose is None else verbose
printed_reasoning = False
printed_text = False
async for event in events:
if isinstance(event, TextDeltaEvent):
if isinstance(event, ReasoningDeltaEvent):
if not printed_reasoning:
click.echo()
click.secho("reasoning: ", fg="bright_black", nl=False)
printed_reasoning = True
_echo_stream(event.content)
elif isinstance(event, TextDeltaEvent):
if printed_reasoning and not printed_text:
click.echo()
if not printed_text:
click.echo()
click.secho("assistant: ", fg="cyan", nl=False)
@@ -59,10 +87,14 @@ async def render_events(events: AsyncIterator[AgentEvent]) -> None: # noqa: C90
else:
click.secho(f"\n[tool] custom id={call.id}", fg="yellow")
elif isinstance(event, ToolResultEvent):
preview = _preview(event.message.content, _TOOL_RESULT_PREVIEW)
# Single-line summary; full result stays in the message history for the model.
one_line = preview.replace("\n", " ")
click.secho(f"[tool ok] {one_line}", fg="magenta")
content = event.message.content
if show_full:
click.secho(f"[tool ok]\n{content}", fg="magenta")
else:
preview = _preview(content, _TOOL_RESULT_PREVIEW)
# Single-line summary; full result stays in the message history for the model.
one_line = preview.replace("\n", " ")
click.secho(f"[tool ok] {one_line}", fg="magenta")
elif isinstance(event, ErrorEvent):
suffix = ""
if event.source:
@@ -89,6 +121,6 @@ async def render_events(events: AsyncIterator[AgentEvent]) -> None: # noqa: C90
else:
# AssistantMessageEvent — text already shown via TextDeltaEvent.
_ = event
if printed_text:
if printed_text or printed_reasoning:
click.echo()
click.echo()
+5 -3
View File
@@ -28,12 +28,14 @@ SLASH_COMMANDS: tuple[str, ...] = (
"/provider",
"/model",
"/tools",
"/stream",
"/verbose",
"/rounds",
"/retry",
"/status",
)
_TOOLS_ARGS: tuple[str, ...] = ("on", "off")
_ON_OFF_ARGS: tuple[str, ...] = ("on", "off")
def history_path() -> Path:
@@ -74,8 +76,8 @@ def _argument_options(state: ReplState, command: str, text: str) -> list[str]:
return filter_prefix(text, sorted(state.config.providers.keys()))
if command == "/model":
return filter_prefix(text, sorted(state.provider.models.keys()))
if command == "/tools":
return filter_prefix(text, list(_TOOLS_ARGS))
if command in {"/tools", "/stream", "/verbose"}:
return filter_prefix(text, list(_ON_OFF_ARGS))
if command == "/resume":
return []
return []
+46 -4
View File
@@ -35,6 +35,8 @@ Commands:
/provider [name] Show or switch provider
/model [id] Show or switch model
/tools [on|off] Show or toggle tools
/stream [on|off] Show or toggle streaming model output
/verbose [on|off] Show or toggle full tool-result dumps
/rounds [n] Show or set max tool-loop rounds
/retry Re-run incomplete last user turn (DB/orphan user; no retype)
/status Show session/provider/tools/rounds status
@@ -43,8 +45,9 @@ User messages are saved immediately. On API errors or Ctrl+C, partial
assistant/tool output is discarded but the user message stays (so /retry
works after resume, not only via readline history). Auto-retry: 10s/20s/30s.
Tab completes slash commands and some arguments (provider, model, tools).
Use --session ID or /resume to continue a prior chat after restart.
Tab completes slash commands and some arguments (provider, model, tools,
stream, verbose). Use --session ID or /resume to continue a prior chat
after restart.
"""
type SlashHandler = Callable[[], None | Awaitable[None]]
@@ -74,7 +77,9 @@ def _cmd_status(state: ReplState) -> None:
f"session={state.session_id} messages={len(state.agent.messages)} "
f"pending_retry={pending_disp}\n"
f"tools={'on' if state.tools_enabled else 'off'} "
f"rounds={state.max_rounds} stream={'on' if state.agent.stream else 'off'}\n"
f"rounds={state.max_rounds} "
f"stream={'on' if state.agent.stream else 'off'} "
f"verbose={'on' if state.verbose else 'off'}\n"
f"context_tokens={ctx_tilde}{ctx_tokens}/{ctx_budget} ({ctx_tag}) "
f"context_chars={ctx_chars} "
f"tool_result_max={state.agent.max_tool_result_chars}\n"
@@ -185,6 +190,40 @@ def _cmd_tools(state: ReplState, arg: str) -> None:
click.echo(f"tools={'on' if state.tools_enabled else 'off'}")
def _cmd_stream(state: ReplState, arg: str) -> None:
token = arg.strip().lower()
if not token:
click.echo(f"stream={'on' if state.agent.stream else 'off'}")
return
if token in {"on", "1", "true", "yes"}:
enabled = True
elif token in {"off", "0", "false", "no"}:
enabled = False
else:
click.echo("usage: /stream [on|off]")
return
state.stream_enabled = enabled
state.agent.stream = enabled
click.echo(f"stream={'on' if enabled else 'off'}")
def _cmd_verbose(state: ReplState, arg: str) -> None:
token = arg.strip().lower()
if not token:
click.echo(f"verbose={'on' if state.verbose else 'off'}")
return
if token in {"on", "1", "true", "yes"}:
enabled = True
elif token in {"off", "0", "false", "no"}:
enabled = False
else:
click.echo("usage: /verbose [on|off]")
return
state.verbose = enabled
state.sync_display_flags()
click.echo(f"verbose={'on' if enabled else 'off'}")
def _cmd_rounds(state: ReplState, arg: str) -> None:
token = arg.strip()
if not token:
@@ -287,6 +326,8 @@ async def _dispatch_slash(state: ReplState, command: str, arg: str) -> bool:
"provider": lambda: _cmd_provider(state, arg),
"model": lambda: _cmd_model(state, arg),
"tools": lambda: _cmd_tools(state, arg),
"stream": lambda: _cmd_stream(state, arg),
"verbose": lambda: _cmd_verbose(state, arg),
"rounds": lambda: _cmd_rounds(state, arg),
"retry": lambda: _cmd_retry(state),
"status": lambda: _cmd_status(state),
@@ -322,7 +363,8 @@ async def run_repl(state: ReplState) -> None:
f"plyngent chat provider={state.provider_name} model={state.model} "
f"session={state.session_id} tools={'on' if state.tools_enabled else 'off'} "
f"rounds={state.max_rounds} messages={len(state.agent.messages)} "
f"stream={'on' if state.agent.stream else 'off'}"
f"stream={'on' if state.agent.stream else 'off'} "
f"verbose={'on' if state.verbose else 'off'}"
)
click.echo("Type /help for commands. Empty line is ignored.")
+13 -1
View File
@@ -29,6 +29,8 @@ class ReplState:
model: str
tools_enabled: bool
max_rounds: int = DEFAULT_MAX_ROUNDS
stream_enabled: bool = True
verbose: bool = False
client: ChatClient = field(init=False)
agent: ChatAgent = field(init=False)
session_id: int | None = None
@@ -38,6 +40,12 @@ class ReplState:
self.client = cast("ChatClient", cast("object", create_client(self.provider)))
self.workspace = Path(self.workspace).expanduser().resolve()
self.agent = self._make_agent()
self.sync_display_flags()
def sync_display_flags(self) -> None:
from plyngent.cli.display import set_verbose_tool_results
set_verbose_tool_results(self.verbose)
def _workspace_key(self) -> str:
key = normalize_workspace(self.workspace)
@@ -74,7 +82,7 @@ class ReplState:
session_id=self.session_id,
max_rounds=self.max_rounds,
on_limit=prompt_continue_limit_async,
stream=True,
stream=self.stream_enabled,
system_prompt=system_prompt,
max_tool_result_chars=agent_cfg.max_tool_result_chars,
parallel_tools=agent_cfg.parallel_tools,
@@ -84,9 +92,13 @@ class ReplState:
def rebuild_client(self) -> None:
"""Recreate client and agent after provider/model/tools change."""
messages = list(self.agent.messages)
# Preserve live stream toggle if agent already exists.
if hasattr(self, "agent"):
self.stream_enabled = self.agent.stream
self.client = cast("ChatClient", cast("object", create_client(self.provider)))
self.agent = self._make_agent()
self.agent.messages = messages
self.sync_display_flags()
def _set_workspace(self, path: Path) -> None:
"""Update REPL + tool workspace root."""
@@ -20,8 +20,8 @@ type DeepSeekReasoningEffort = ReasoningEffort | Literal["max"]
class AssistantChatMessage(BaseAssistantChatMessage):
# reasoning_content lives on the base assistant message (OpenAI-compat).
prefix: bool | Unset = UNSET
reasoning_content: str | Unset = UNSET
class ToolChatMessage(BaseToolChatMessage):
@@ -72,6 +72,8 @@ class AssistantChatMessage(Struct, tag_field="role", tag="assistant"):
audio: IDObject | Unset = UNSET
refusal: str | Unset = UNSET
tool_calls: list[AnyAssistantToolCall] | Unset = UNSET
# OpenAI-compat / DeepSeek thinking streams (omitted when unset).
reasoning_content: str | Unset = UNSET
class ToolChatMessage(ChatMessage, tag_field="role", tag="tool"):
@@ -235,6 +237,7 @@ class StreamToolCallDelta(Struct):
class DeltaMessage(Struct):
role: RoleAssistant | Unset = UNSET
content: str | Unset = UNSET
reasoning_content: str | Unset = UNSET
tool_calls: list[StreamToolCallDelta] | Unset = UNSET
+73
View File
@@ -9,6 +9,7 @@ from plyngent.agent import (
AssistantMessageEvent,
ChatAgent,
MaxRoundsEvent,
ReasoningDeltaEvent,
TextDeltaEvent,
ToolCallEvent,
ToolRegistry,
@@ -338,6 +339,78 @@ async def test_stream_yields_deltas_incrementally() -> None:
assert messages[-1].content == "ab"
async def test_stream_yields_reasoning_deltas() -> None:
class ReasoningClient:
@overload
async def chat_completions(
self, param: ChatCompletionsParam, *, stream: Literal[False] = False
) -> ChatCompletionResponse: ...
@overload
async def chat_completions(
self, param: ChatCompletionsParam, *, stream: Literal[True]
) -> AsyncIterator[ChatCompletionChunk]: ...
async def chat_completions(
self, param: ChatCompletionsParam, *, stream: bool = False
) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]:
del param
if not stream:
return _response(AssistantChatMessage(content="ans", reasoning_content="think"))
async def chunks() -> AsyncIterator[ChatCompletionChunk]:
for part in ("th", "ink"):
yield ChatCompletionChunk(
id="1",
object="chat.completion.chunk",
created=0,
model="t",
choices=[
ChunkChoice(
index=0,
delta=DeltaMessage(reasoning_content=part),
finish_reason=None,
)
],
)
yield ChatCompletionChunk(
id="1",
object="chat.completion.chunk",
created=0,
model="t",
choices=[
ChunkChoice(
index=0,
delta=DeltaMessage(content="ans"),
finish_reason="stop",
)
],
)
return chunks()
messages: list[AnyChatMessage] = [UserChatMessage(content="hi")]
events = [e async for e in run_chat_loop(ReasoningClient(), messages, model="m", stream=True)]
reasoning = [e for e in events if isinstance(e, ReasoningDeltaEvent)]
assert [r.content for r in reasoning] == ["th", "ink"]
deltas = [e for e in events if isinstance(e, TextDeltaEvent)]
assert [d.content for d in deltas] == ["ans"]
assert isinstance(messages[-1], AssistantChatMessage)
assert messages[-1].content == "ans"
assert messages[-1].reasoning_content == "think"
async def test_non_stream_yields_reasoning() -> None:
client = ScriptedClient([_response(AssistantChatMessage(content="ok", reasoning_content="plan"))])
messages: list[AnyChatMessage] = [UserChatMessage(content="hi")]
events = [e async for e in run_chat_loop(client, messages, model="m", stream=False)]
reasoning = [e for e in events if isinstance(e, ReasoningDeltaEvent)]
assert len(reasoning) == 1
assert reasoning[0].content == "plan"
assert isinstance(messages[-1], AssistantChatMessage)
assert messages[-1].reasoning_content == "plan"
async def test_run_chat_loop_with_tools() -> None:
@tool
def add(a: int, b: int) -> int:
+51
View File
@@ -0,0 +1,51 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from plyngent.agent import ReasoningDeltaEvent, TextDeltaEvent, ToolResultEvent
from plyngent.cli.display import render_events, set_verbose_tool_results
from plyngent.lmproto.openai_compatible.model import ToolChatMessage
if TYPE_CHECKING:
from collections.abc import AsyncIterator
import pytest
from plyngent.agent import AgentEvent
async def _aiter(events: list[AgentEvent]) -> AsyncIterator[AgentEvent]:
for event in events:
yield event
async def test_render_reasoning_and_text(capsys: pytest.CaptureFixture[str]) -> None:
await render_events(
_aiter(
[
ReasoningDeltaEvent(content="think"),
TextDeltaEvent(content="hello"),
]
)
)
out = capsys.readouterr().out
assert "reasoning: " in out
assert "think" in out
assert "assistant: " in out
assert "hello" in out
async def test_tool_result_preview_vs_verbose(capsys: pytest.CaptureFixture[str]) -> None:
long = "x" * 200
msg = ToolChatMessage(content=long, tool_call_id="1")
set_verbose_tool_results(False)
await render_events(_aiter([ToolResultEvent(message=msg)]))
out = capsys.readouterr().out
assert "" in out
assert long not in out
set_verbose_tool_results(True)
await render_events(_aiter([ToolResultEvent(message=msg)]), verbose=True)
out2 = capsys.readouterr().out
assert long in out2
set_verbose_tool_results(False)
+2
View File
@@ -77,6 +77,8 @@ def test_completer_commands(tmp_path: object, monkeypatch: object) -> None:
found.append(item)
index += 1
assert "/help" in found
assert "/stream" in SLASH_COMMANDS
assert "/verbose" in SLASH_COMMANDS
assert set(found) <= set(SLASH_COMMANDS)
+21
View File
@@ -116,6 +116,27 @@ async def test_tools_toggle(state: ReplState) -> None:
assert state.tools_enabled is False
async def test_stream_toggle(state: ReplState) -> None:
assert state.agent.stream is True
assert await handle_slash(state, "/stream off") is True
assert state.agent.stream is False
assert state.stream_enabled is False
assert await handle_slash(state, "/stream on") is True
assert state.agent.stream is True
async def test_verbose_toggle(state: ReplState) -> None:
from plyngent.cli.display import get_verbose_tool_results
assert state.verbose is False
assert await handle_slash(state, "/verbose on") is True
assert state.verbose is True
assert get_verbose_tool_results() is True
assert await handle_slash(state, "/verbose off") is True
assert state.verbose is False
assert get_verbose_tool_results() is False
async def test_resume(state: ReplState) -> None:
sid = state.session_id
assert sid is not None