From fe37f8be5c0cdac484a09f913050c919c4f08c8f Mon Sep 17 00:00:00 2001 From: renjianbo <263303411@qq.com> Date: Sun, 26 Jul 2026 20:57:11 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20AgentChat=20streaming=20UX=20=E2=80=94?= =?UTF-8?q?=20answer=5Fchunk=20typewriter,=20token/cost=20display,=20stop?= =?UTF-8?q?=20marker,=20session=20manage,=20date=20dividers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Backend: token-level streaming via on_delta callback + asyncio.Queue bridge, answer_chunk SSE events; final event carries token_usage with cost_yuan; kimi pricing in cost_estimator - Frontend: typewriter rendering of answer chunks; token/cost in message meta; abort marks message '已停止' and no longer triggers duplicate non-stream fallback; session pin/delete in dropdown (pinned first); date dividers across days --- backend/app/agent_runtime/core.py | 124 ++++++++++- backend/app/services/cost_estimator.py | 2 + frontend/src/components/ChatMessageBubble.vue | 12 ++ frontend/src/views/AgentChat.vue | 193 ++++++++++++++---- 4 files changed, 284 insertions(+), 47 deletions(-) diff --git a/backend/app/agent_runtime/core.py b/backend/app/agent_runtime/core.py index 5950430..a0a8d6d 100644 --- a/backend/app/agent_runtime/core.py +++ b/backend/app/agent_runtime/core.py @@ -14,6 +14,7 @@ import hashlib import json import logging import time +import asyncio from typing import Any, AsyncGenerator, Callable, Dict, List, Optional, Protocol, TypedDict from app.agent_runtime.schemas import ( @@ -271,13 +272,23 @@ class AgentRuntime: def _real_token_usage_payload(self) -> Dict[str, Any]: """构造随终态事件下发的真实用量 dict（形状含 total_tokens，供下游/前端与计费消费）。""" - return { + payload = { "prompt_tokens": self._real_usage["prompt_tokens"], "completion_tokens": self._real_usage["completion_tokens"], "total_tokens": self._real_usage["total_tokens"], "llm_calls": self._real_usage["llm_calls"], "model": self.config.llm.model, } + try: + from app.services.cost_estimator import estimate_cost_yuan + payload["cost_yuan"] = estimate_cost_yuan( + payload["model"] or "", + payload["prompt_tokens"], + payload["completion_tokens"], + ) + except Exception: + pass + return payload def _attach_token_usage(self, result: AgentResult) -> AgentResult: """将 token 用量摘要附加到 AgentResult。预算启用则用其摘要；否则用真实累计用量。""" @@ -978,15 +989,36 @@ class AgentRuntime: # think 事件：告知前端 Agent 正在思考（让 UI 即时反馈，避免假死感） yield {"type": "think", "content": "", "reasoning": None, "iteration": self.context.iteration} - # 调用 LLM + # 调用 LLM（token 级流式：delta 进队列，边下边 yield answer_chunk） try: - response = await llm.chat( + delta_q: asyncio.Queue = asyncio.Queue() + + def _on_delta(text: str) -> None: + try: + delta_q.put_nowait(text) + except Exception: + pass + + chat_task = asyncio.create_task(llm.chat( messages=messages, tools=tool_schemas if has_tools and self.context.iteration == 1 else (tool_schemas if has_tools else None), iteration=self.context.iteration, on_completion=_llm_callback, - ) + on_delta=_on_delta, + )) + while True: + try: + piece = await asyncio.wait_for(delta_q.get(), timeout=0.2) + yield {"type": "answer_chunk", "content": piece, + "iteration": self.context.iteration} + except asyncio.TimeoutError: + if chat_task.done(): + break + while not delta_q.empty(): + yield {"type": "answer_chunk", "content": delta_q.get_nowait(), + "iteration": self.context.iteration} + response = chat_task.result() except Exception as e: err_str = str(e) logger.error("LLM 调用失败 (iteration=%s): %s", self.context.iteration, err_str) @@ -1803,8 +1835,10 @@ class _LLMClient: tools: Optional[List[Dict[str, Any]]] = None, iteration: int = 1, on_completion: Optional[Callable[[Dict[str, Any]], Any]] = None, + on_delta: Optional[Callable[[str], None]] = None, ) -> Any: - """调用 LLM，主模型失败时自动切换 fallback_llm 重试。""" + """调用 LLM，主模型失败时自动切换 fallback_llm 重试。 + on_delta 提供时启用 token 级流式（仅内容增量，工具调用仍会重组后返回）。""" from openai import AsyncOpenAI from app.core.config import settings @@ -1822,7 +1856,7 @@ class _LLMClient: return await self._do_chat( api_key=api_key, base_url=base_url, model=self._config.model, messages=messages, tools=tools, iteration=iteration, - on_completion=on_completion, + on_completion=on_completion, on_delta=on_delta, ) async def _do_chat( @@ -1835,6 +1869,7 @@ class _LLMClient: iteration: int = 1, on_completion: Optional[Callable[[Dict[str, Any]], Any]] = None, _is_fallback: bool = False, + on_delta: Optional[Callable[[str], None]] = None, ) -> Any: from openai import AsyncOpenAI from app.core.config import settings @@ -1883,10 +1918,22 @@ class _LLMClient: tool_calls = None return _CachedMsg() + # token 级流式：要求增量回调时走 stream=True，末 chunk 带 usage + if on_delta: + kwargs["stream"] = True + kwargs["stream_options"] = {"include_usage": True} + start_time = time.perf_counter() last_error = None + message = None + usage = None try: - response = await client.chat.completions.create(**kwargs) + if on_delta: + message, usage = await self._stream_chat(client, kwargs, on_delta) + else: + response = await client.chat.completions.create(**kwargs) + message = response.choices[0].message + usage = getattr(response, "usage", None) except Exception as e: last_error = e @@ -1914,6 +1961,7 @@ class _LLMClient: iteration=iteration, on_completion=on_completion, _is_fallback=_is_fallback, + on_delta=on_delta, ) except Exception as ce: logger.error("ReactiveCompact 失败: %s", ce) @@ -1953,12 +2001,11 @@ class _LLMClient: model=fb_model, messages=messages, tools=tools, iteration=iteration, on_completion=on_completion, - _is_fallback=True, + _is_fallback=True, on_delta=on_delta, ) raise latency_ms = int((time.perf_counter() - start_time) * 1000) - message = response.choices[0].message # 缓存写入（仅不用工具时） if self._config.cache_enabled and not tools and message.content: @@ -1966,7 +2013,6 @@ class _LLMClient: await _llm_cache_set(ck, message.content, self._config.cache_ttl_ms) # 提取 token 用量 - usage = getattr(response, "usage", None) prompt_tokens = usage.prompt_tokens if usage else 0 completion_tokens = usage.completion_tokens if usage else 0 total_tokens = usage.total_tokens if usage else 0 @@ -1985,3 +2031,61 @@ class _LLMClient: }) return message + + @staticmethod + async def _stream_chat(client: Any, kwargs: Dict[str, Any], on_delta: Callable[[str], None]) -> Any: + """消费 OpenAI 兼容流式响应：逐 delta 回调 on_delta，重组为 message 对象返回。 + + 返回 (message, usage)：message 为 SimpleNamespace（content/tool_calls/ + reasoning_content，duck-type 兼容 SDK Message），usage 来自开启 + stream_options.include_usage 后的末 chunk。 + """ + from types import SimpleNamespace + + stream = await client.chat.completions.create(**kwargs) + content_parts: List[str] = [] + reasoning_parts: List[str] = [] + tool_slots: Dict[int, Dict[str, Any]] = {} + usage = None + + async for chunk in stream: + if getattr(chunk, "usage", None): + usage = chunk.usage + if not chunk.choices: + continue + delta = chunk.choices[0].delta + piece = getattr(delta, "content", None) + if piece: + content_parts.append(piece) + try: + on_delta(piece) + except Exception: + pass + reasoning_piece = getattr(delta, "reasoning_content", None) + if reasoning_piece: + reasoning_parts.append(reasoning_piece) + for tc in (getattr(delta, "tool_calls", None) or []): + slot = tool_slots.setdefault(tc.index, {"id": None, "name": "", "arguments": ""}) + if tc.id: + slot["id"] = tc.id + fn = getattr(tc, "function", None) + if fn is not None: + if getattr(fn, "name", None): + slot["name"] += fn.name + if getattr(fn, "arguments", None): + slot["arguments"] += fn.arguments + + tool_calls = [ + SimpleNamespace( + id=s["id"], type="function", + function=SimpleNamespace(name=s["name"], arguments=s["arguments"]), + ) + for _, s in sorted(tool_slots.items()) + ] or None + + message = SimpleNamespace( + content="".join(content_parts) if content_parts else None, + tool_calls=tool_calls, + reasoning_content="".join(reasoning_parts) if reasoning_parts else None, + ) + return message, usage diff --git a/backend/app/services/cost_estimator.py b/backend/app/services/cost_estimator.py index 8b68c18..312a20d 100644 --- a/backend/app/services/cost_estimator.py +++ b/backend/app/services/cost_estimator.py @@ -23,6 +23,8 @@ MODEL_PRICING: Dict[str, Tuple[float, float]] = { "claude-3-haiku": (0.25, 1.25), "claude-3.5-sonnet": (3.00, 15.00), "claude-3.5-haiku": (1.00, 5.00), + "kimi-k3": (3.00, 15.00), + "kimi-k2": (0.56, 2.22), } # 缓存未命中模型的默认定价 diff --git a/frontend/src/components/ChatMessageBubble.vue b/frontend/src/components/ChatMessageBubble.vue index b5c9b9d..2a5e560 100644 --- a/frontend/src/components/ChatMessageBubble.vue +++ b/frontend/src/components/ChatMessageBubble.vue @@ -32,7 +32,9 @@
{{ formatTime(timestamp) }} + 已停止 {{ iterations }} 步 · {{ toolCallsMade }} 次工具调用 + {{ tokenUsageText }} @@ -54,6 +56,7 @@ const props = defineProps<{ attachments?: MessageAttachment[] iterations?: number toolCallsMade?: number + tokenUsage?: any }>() defineEmits<{ @@ -62,6 +65,14 @@ defineEmits<{ const attachmentsClass = computed(() => (props.role === 'user' ? 'is-user' : 'is-agent')) +const tokenUsageText = computed(() => { + const u = props.tokenUsage + if (!u || !u.total_tokens) return '' + let text = `${u.total_tokens} tokens` + if (u.cost_yuan !== undefined && u.cost_yuan !== null) text += ` · ¥${u.cost_yuan}` + return text +}) + function formatTime(ts: number): string { const d = new Date(ts) return d.toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit' }) @@ -194,6 +205,7 @@ function formatTime(ts: number): string { flex-wrap: wrap; } .msg-iterations { color: var(--el-color-info); } +.msg-token-usage { color: var(--el-text-color-placeholder); } .msg-actions { margin-left: auto; display: flex; diff --git a/frontend/src/views/AgentChat.vue b/frontend/src/views/AgentChat.vue index 9063af8..082cb45 100644 --- a/frontend/src/views/AgentChat.vue +++ b/frontend/src/views/AgentChat.vue @@ -33,14 +33,33 @@ + 新对话 -
- {{ s.title || '未命名对话' }} - {{ s.message_count }} 条消息 · {{ relativeTimeText(s.updated_at) }} +
+
+ + + {{ s.title || '未命名对话' }} + + {{ s.message_count }} 条消息 · {{ relativeTimeText(s.updated_at) }} +
+ + + +
@@ -170,24 +189,26 @@
- -