From 4ae8262aa6ec0425bcd3a964d2f94182584fe783 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Sun, 25 Oct 2026 16:44:30 +0800 Subject: [PATCH] feat(activity): add SAS stream_delta API for LLM progressive text (1.2.14) --- README.md | 17 ++ pyproject.toml | 2 +- src/jiangchang_skill_core/__init__.py | 10 ++ src/jiangchang_skill_core/activity.py | 227 +++++++++++++++++++++++++- tests/test_activity_stream.py | 144 ++++++++++++++++ 5 files changed, 398 insertions(+), 2 deletions(-) create mode 100644 tests/test_activity_stream.py diff --git a/README.md b/README.md index ace205a..2cce09e 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,8 @@ send_prompt_via_composer(page, "第一行\n第二行") |-----|------| | `emit(text, ...)` | 写 Run Journal（`activity` / `progress` / `warn` / `debug`） | | `step(text)` | `emit(type='activity')` 别名 | +| `stream_start` / `stream_delta` / `stream_end` | 写 `type=stream` 渐进文本（**不**步进闸门；供 HTML `jiangchang.stream`） | +| `stream_channel(...)` | 上下文管理器：start → delta → end | | `finish(status=..., **fields)` | 写终态事件 + **唯一**向 stdout 输出单行 result JSON | | `rpa_step(name)` | 装饰器：闸门 + 进入/成功/失败自动 emit | | `checkpoint()` | consult control（不写步骤事件） | @@ -94,6 +96,21 @@ send_prompt_via_composer(page, "第一行\n第二行") | `JobStopped` | 宿主 stop 时抛出 | | `job_context(...)` | 可选包裹 main；异常时自动 `finish(failed)` | +### LLM 流式示例（1.2.14+） + +```python +from jiangchang_skill_core.activity import finish, stream_channel + +with stream_channel("answer", skill="reply-xmz-knowledge") as ch: + for piece in llm_stream(...): # OpenAI/DeepSeek stream chunks + ch.delta(piece) + +finish(status="success", skill="reply-xmz-knowledge", message=full_text, answer=full_text) +``` + +- `stream_delta` 默认合并短突发（约 50ms / 48 字符）；可用环境变量 `JIANGCHANG_STREAM_COALESCE_MS` / `JIANGCHANG_STREAM_COALESCE_CHARS` 调整，或 `flush=True` 立即写出。 +- 宿主 Task Center **不展示**逐字 `stream` delta；HTML 预览通过 `jiangchang.stream` 订阅。 + ### Run Journal 路径（宿主 tail 此文件） ``` diff --git a/pyproject.toml b/pyproject.toml index a68b3f8..d158f34 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "1.2.13" +version = "1.2.14" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK + 桌面 UIA 暖树" readme = "README.md" requires-python = ">=3.10" diff --git a/src/jiangchang_skill_core/__init__.py b/src/jiangchang_skill_core/__init__.py index b86ce2e..95f1684 100644 --- a/src/jiangchang_skill_core/__init__.py +++ b/src/jiangchang_skill_core/__init__.py @@ -1,5 +1,6 @@ from .activity import ( JobStopped, + StreamChannel, checkpoint, emit, finish, @@ -11,6 +12,10 @@ from .activity import ( job_context, rpa_step, step, + stream_channel, + stream_delta, + stream_end, + stream_start, ) from .sibling_bridge import ( call_sibling_json, @@ -74,6 +79,7 @@ except ImportError: __all__ = [ "JobStopped", + "StreamChannel", "call_sibling_json", "call_sibling_json_array", "checkpoint", @@ -89,6 +95,10 @@ __all__ = [ "job_context", "rpa_step", "step", + "stream_channel", + "stream_delta", + "stream_end", + "stream_start", "get_sibling_main_path", "EntitlementClient", "EntitlementDeniedError", diff --git a/src/jiangchang_skill_core/activity.py b/src/jiangchang_skill_core/activity.py index 60a3718..e5c6d71 100644 --- a/src/jiangchang_skill_core/activity.py +++ b/src/jiangchang_skill_core/activity.py @@ -4,6 +4,9 @@ Skill Activity Stream (SAS) v1 + Skill Runtime Control Protocol (SRCP) v1. emit() / journal events never touch stdout (Agent/bash tool result stays clean). finish() is the only API that writes a single-line result JSON to stdout. +Stream helpers (stream_start / stream_delta / stream_end / stream_channel) write +type=stream journal events without advancing the step gate — for LLM token UX. + SRCP: host writes {job_id}.control.json; kit consults at step gates (emit, rpa_step, interruptible_sleep) for pause / resume / stop. """ @@ -26,9 +29,13 @@ from .runtime_env import ensure_runs_dir, get_runs_dir _SCHEMA_VERSION = 1 _CONTROL_SCHEMA_VERSION = 1 _VALID_EMIT_TYPES = frozenset({"activity", "progress", "warn", "debug"}) +_VALID_STREAM_PHASES = frozenset({"start", "delta", "end"}) _VALID_CONTROL_COMMANDS = frozenset({"none", "pause", "resume", "stop"}) _RESULT_MAX_CHARS = 4096 _DEFAULT_CONTROL_POLL_SEC = 0.2 +_DEFAULT_STREAM_COALESCE_MS = 50 +_DEFAULT_STREAM_COALESCE_CHARS = 48 +_STREAM_CHECKPOINT_EVERY_MS = 200 _cached_job_id: Optional[str] = None _journal_path: Optional[str] = None @@ -36,6 +43,8 @@ _meta_written: bool = False _journal_file: Optional[Any] = None _step_index: int = 0 _current_step_name: Optional[str] = None +_stream_last_checkpoint_ms: int = 0 +_stream_buffers: dict[str, dict[str, Any]] = {} F = TypeVar("F", bound=Callable[..., Any]) @@ -320,6 +329,220 @@ def step(text: str, **kwargs: Any) -> None: emit(text, type="activity", **kwargs) +def _stream_coalesce_ms() -> int: + raw = (os.getenv("JIANGCHANG_STREAM_COALESCE_MS") or "").strip() + if not raw: + return _DEFAULT_STREAM_COALESCE_MS + try: + return max(0, int(raw)) + except ValueError: + return _DEFAULT_STREAM_COALESCE_MS + + +def _stream_coalesce_chars() -> int: + raw = (os.getenv("JIANGCHANG_STREAM_COALESCE_CHARS") or "").strip() + if not raw: + return _DEFAULT_STREAM_COALESCE_CHARS + try: + return max(1, int(raw)) + except ValueError: + return _DEFAULT_STREAM_COALESCE_CHARS + + +def _maybe_stream_checkpoint() -> None: + """Consult stop/pause without advancing step index (throttled).""" + global _stream_last_checkpoint_ms + now = _now_ms() + if now - _stream_last_checkpoint_ms < _STREAM_CHECKPOINT_EVERY_MS: + return + _stream_last_checkpoint_ms = now + checkpoint() + + +def _write_stream_event( + *, + phase: str, + stream_id: str, + text: str, + delta: str | None = None, + skill: str | None = None, + **extra: Any, +) -> None: + if phase not in _VALID_STREAM_PHASES: + raise ValueError(f"invalid stream phase: {phase!r}") + sid = (stream_id or "answer").strip() or "answer" + label = (text or "").strip() or sid + event = _build_event( + "stream", + label, + skill=skill, + phase=phase, + stream_id=sid, + **extra, + ) + if delta is not None: + event["delta"] = delta + _write_journal_line(event) + + +def stream_start( + stream_id: str = "answer", + *, + skill: str | None = None, + text: str | None = None, + **extra: Any, +) -> None: + """Open a progressive text stream (no step gate).""" + _maybe_stream_checkpoint() + sid = (stream_id or "answer").strip() or "answer" + buf = _stream_buffers.get(sid) + if buf and buf.get("pending"): + _flush_stream_buffer(sid, force=True) + _stream_buffers[sid] = { + "pending": "", + "last_flush_ms": _now_ms(), + "skill": skill, + "extra": dict(extra), + } + _write_stream_event( + phase="start", + stream_id=sid, + text=text or f"stream:{sid}", + skill=skill, + **extra, + ) + + +def _flush_stream_buffer(stream_id: str, *, force: bool = False) -> None: + buf = _stream_buffers.get(stream_id) + if not buf: + return + pending = buf.get("pending") or "" + if not pending: + return + coalesce_ms = _stream_coalesce_ms() + coalesce_chars = _stream_coalesce_chars() + now = _now_ms() + aged = now - int(buf.get("last_flush_ms") or 0) >= coalesce_ms + big = len(pending) >= coalesce_chars + if not force and not aged and not big: + return + _maybe_stream_checkpoint() + skill = buf.get("skill") + extra = dict(buf.get("extra") or {}) + _write_stream_event( + phase="delta", + stream_id=stream_id, + text=pending, + delta=pending, + skill=skill if isinstance(skill, str) else None, + **extra, + ) + buf["pending"] = "" + buf["last_flush_ms"] = now + + +def stream_delta( + delta: str, + *, + stream_id: str = "answer", + skill: str | None = None, + flush: bool = False, + **extra: Any, +) -> None: + """ + Append progressive text to the Run Journal as type=stream phase=delta. + + Does not advance the SAS step index. Empty deltas are ignored. + Small chunks are coalesced (~50ms / ~48 chars) unless flush=True. + """ + piece = delta if isinstance(delta, str) else str(delta or "") + if piece == "": + return + sid = (stream_id or "answer").strip() or "answer" + buf = _stream_buffers.get(sid) + if buf is None: + stream_start(sid, skill=skill, **extra) + buf = _stream_buffers.get(sid) + assert buf is not None + if skill is not None: + buf["skill"] = skill + if extra: + merged = dict(buf.get("extra") or {}) + merged.update(extra) + buf["extra"] = merged + buf["pending"] = (buf.get("pending") or "") + piece + _flush_stream_buffer(sid, force=flush) + + +def stream_end( + stream_id: str = "answer", + *, + skill: str | None = None, + text: str | None = None, + **extra: Any, +) -> None: + """Flush pending deltas and close a progressive text stream.""" + sid = (stream_id or "answer").strip() or "answer" + _flush_stream_buffer(sid, force=True) + buf = _stream_buffers.pop(sid, None) + use_skill = skill + if use_skill is None and buf and isinstance(buf.get("skill"), str): + use_skill = buf["skill"] + merged_extra = dict(buf.get("extra") or {}) if buf else {} + merged_extra.update(extra) + _maybe_stream_checkpoint() + _write_stream_event( + phase="end", + stream_id=sid, + text=text or f"stream:{sid}:end", + skill=use_skill, + **merged_extra, + ) + + +class StreamChannel: + """Context manager around stream_start / stream_delta / stream_end.""" + + def __init__( + self, + stream_id: str = "answer", + *, + skill: str | None = None, + **extra: Any, + ) -> None: + self.stream_id = (stream_id or "answer").strip() or "answer" + self.skill = skill + self.extra = dict(extra) + + def __enter__(self) -> "StreamChannel": + stream_start(self.stream_id, skill=self.skill, **self.extra) + return self + + def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None: + stream_end(self.stream_id, skill=self.skill) + return None + + def delta(self, text: str, *, flush: bool = False) -> None: + stream_delta( + text, + stream_id=self.stream_id, + skill=self.skill, + flush=flush, + **self.extra, + ) + + +def stream_channel( + stream_id: str = "answer", + *, + skill: str | None = None, + **extra: Any, +) -> StreamChannel: + """Return a StreamChannel context manager for progressive LLM text.""" + return StreamChannel(stream_id, skill=skill, **extra) + + async def interruptible_sleep(seconds: float, *, chunk: float | None = None) -> None: """asyncio.sleep in slices; each slice consults host control (pause/stop).""" if seconds <= 0: @@ -532,7 +755,7 @@ def rpa_step(name: str | None = None) -> Callable[[F], F]: def _reset_for_tests() -> None: """Clear module-level journal and control state (tests only).""" global _cached_job_id, _journal_path, _meta_written, _journal_file - global _step_index, _current_step_name + global _step_index, _current_step_name, _stream_last_checkpoint_ms, _stream_buffers if _journal_file is not None: try: _journal_file.close() @@ -544,3 +767,5 @@ def _reset_for_tests() -> None: _journal_file = None _step_index = 0 _current_step_name = None + _stream_last_checkpoint_ms = 0 + _stream_buffers = {} diff --git a/tests/test_activity_stream.py b/tests/test_activity_stream.py new file mode 100644 index 0000000..92f04e7 --- /dev/null +++ b/tests/test_activity_stream.py @@ -0,0 +1,144 @@ +# -*- coding: utf-8 -*- +"""SAS stream_start / stream_delta / stream_end / stream_channel.""" +from __future__ import annotations + +import json +import os +import shutil +import tempfile +import unittest +from pathlib import Path + +from jiangchang_skill_core import activity +from jiangchang_skill_core.runtime_env import get_runs_dir + + +class _ActivityEnvIsolation(unittest.TestCase): + _KEYS = ( + "JIANGCHANG_DATA_ROOT", + "JIANGCHANG_USER_ID", + "JIANGCHANG_JOB_ID", + "JIANGCHANG_RUN_MODE", + "JIANGCHANG_ACTIVITY_DEBUG", + "JIANGCHANG_STREAM_COALESCE_MS", + "JIANGCHANG_STREAM_COALESCE_CHARS", + "JIANGCHANG_CONTROL_DISABLED", + ) + + def setUp(self) -> None: + self._saved = {k: os.environ.get(k) for k in self._KEYS} + self._tmps: list[str] = [] + for k in self._KEYS: + os.environ.pop(k, None) + os.environ["JIANGCHANG_CONTROL_DISABLED"] = "1" + activity._reset_for_tests() + + def tearDown(self) -> None: + activity._reset_for_tests() + for k, v in self._saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + for tmp in self._tmps: + shutil.rmtree(tmp, ignore_errors=True) + + def _tmp_data_root(self) -> str: + tmp = tempfile.mkdtemp() + self._tmps.append(tmp) + return tmp + + def _events(self, job_id: str) -> list[dict]: + journal = Path(get_runs_dir()) / f"{job_id}.jsonl" + lines = journal.read_text(encoding="utf-8").strip().splitlines() + return [json.loads(line) for line in lines] + + +class TestActivityStream(_ActivityEnvIsolation): + def test_stream_writes_start_delta_end_without_step_gate(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-stream-1" + os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0" + activity._reset_for_tests() + + activity.stream_start("answer", skill="demo") + activity.stream_delta("你", stream_id="answer", skill="demo") + activity.stream_delta("好", stream_id="answer", skill="demo") + activity.stream_end("answer", skill="demo") + + events = self._events("job-stream-1") + stream_events = [e for e in events if e.get("type") == "stream"] + phases = [e.get("phase") for e in stream_events] + self.assertEqual(phases[0], "start") + self.assertEqual(phases[-1], "end") + deltas = [e.get("delta") for e in stream_events if e.get("phase") == "delta"] + self.assertEqual("".join(deltas), "你好") + self.assertEqual(activity._step_index, 0) + + def test_stream_channel_context_manager(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-stream-ch" + os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0" + activity._reset_for_tests() + + with activity.stream_channel("answer", skill="demo") as ch: + ch.delta("A") + ch.delta("B", flush=True) + + events = self._events("job-stream-ch") + stream_events = [e for e in events if e.get("type") == "stream"] + self.assertEqual(stream_events[0]["phase"], "start") + self.assertEqual(stream_events[-1]["phase"], "end") + deltas = [e.get("delta") for e in stream_events if e.get("phase") == "delta"] + self.assertEqual("".join(d for d in deltas if d), "AB") + + def test_stream_delta_auto_starts(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-stream-auto" + os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0" + activity._reset_for_tests() + + activity.stream_delta("x", stream_id="answer") + activity.stream_end("answer") + + events = self._events("job-stream-auto") + stream_events = [e for e in events if e.get("type") == "stream"] + self.assertEqual(stream_events[0]["phase"], "start") + self.assertTrue(any(e.get("phase") == "delta" and e.get("delta") == "x" for e in stream_events)) + + def test_empty_delta_ignored(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-stream-empty" + os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0" + activity._reset_for_tests() + + activity.stream_start("answer") + activity.stream_delta("", stream_id="answer") + activity.stream_end("answer") + + events = self._events("job-stream-empty") + stream_events = [e for e in events if e.get("type") == "stream"] + self.assertFalse(any(e.get("phase") == "delta" for e in stream_events)) + + def test_public_exports(self) -> None: + from jiangchang_skill_core import ( + StreamChannel, + stream_channel, + stream_delta, + stream_end, + stream_start, + ) + + self.assertTrue(callable(stream_start)) + self.assertTrue(callable(stream_delta)) + self.assertTrue(callable(stream_end)) + self.assertTrue(callable(stream_channel)) + self.assertTrue(issubclass(StreamChannel, object)) + + +if __name__ == "__main__": + unittest.main()