From 83486de1d129ebb1b501de368f43d9023979ee0d Mon Sep 17 00:00:00 2001
From: Hengjie Wang <123135745+anchorAnc@users.noreply.github.com>
Date: Fri, 1 May 2026 18:17:43 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E5=8F=AF=E9=80=89?=
 =?UTF-8?q?=E7=9A=84=E6=B5=81=E5=BC=8F=20LLM=20=E8=B0=83=E7=94=A8=20(#181?=
 =?UTF-8?q?=20@anchorAnc)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* feat: add optional streaming LLM provider call

* fix(llm_utils): code quality

---------

Co-authored-by: SXP-Simon <sxp20061207@163.com>
---
 _conf_schema.json                             |  8 ++-
 .../analysis/utils/llm_utils.py               | 55 +++++++++++++++++--
 src/infrastructure/config/config_manager.py   |  4 ++
 3 files changed, 60 insertions(+), 7 deletions(-)

diff --git a/_conf_schema.json b/_conf_schema.json
index 0e99f6e..31d3ad1 100644
--- a/_conf_schema.json
+++ b/_conf_schema.json
@@ -337,6 +337,12 @@
                 "description": "LLM 请求重试退避基值（秒）",
                 "default": 2,
                 "hint": "重试之间的基准等待时间（秒），实际等待时间为基值乘以尝试次数。"
+            },
+            "enable_streaming_llm_call": {
+                "type": "bool",
+                "description": "启用流式 LLM 调用",
+                "default": false,
+                "hint": "默认关闭，使用原有非流式调用方式。开启后，插件将使用 AstrBot Provider 的流式接口并聚合结果，适用于仅支持 stream=true 的服务。"
             }
         }
     },
@@ -669,4 +675,4 @@
             }
         }
     }
-}
\ No newline at end of file
+}
diff --git a/src/infrastructure/analysis/utils/llm_utils.py b/src/infrastructure/analysis/utils/llm_utils.py
index 018e68a..ecece36 100644
--- a/src/infrastructure/analysis/utils/llm_utils.py
+++ b/src/infrastructure/analysis/utils/llm_utils.py
@@ -5,6 +5,8 @@ LLM API请求处理工具模块
 
 import asyncio
 
+from astrbot.api.provider import LLMResponse
+
 from ....utils.logger import logger
 from ....utils.resilience import CircuitBreaker, GlobalRateLimiter
 from .structured_output_schema import JSONObject
@@ -36,6 +38,40 @@ def _get_circuit_breaker(provider_id: str) -> CircuitBreaker:
     return _circuit_breakers[provider_id]
 
 
+async def _call_provider_stream(
+    context, provider_id: str, llm_kwargs: dict[str, object]
+):
+    provider = context.get_provider_by_id(provider_id=provider_id)
+    if provider is None:
+        raise RuntimeError(f"Provider 不存在: {provider_id}")
+
+    stream_kwargs = dict(llm_kwargs)
+    stream_kwargs.pop("chat_provider_id", None)
+
+    final_resp = None
+    content_parts: list[str] = []
+    async for resp in provider.text_chat_stream(**stream_kwargs):
+        final_resp = resp
+        if getattr(resp, "is_chunk", False):
+            text = getattr(resp, "completion_text", "")
+            if text:
+                content_parts.append(text)
+
+    if final_resp is None:
+        raise RuntimeError("流式 LLM 调用未返回任何响应")
+
+    final_text = extract_response_text(final_resp)
+    if final_text and not getattr(final_resp, "is_chunk", False):
+        return final_resp
+
+    return LLMResponse(
+        role="assistant",
+        completion_text="".join(content_parts),
+        usage=getattr(final_resp, "usage", None),
+        raw_completion=getattr(final_resp, "raw_completion", None),
+    )
+
+
 async def _try_get_provider_id_by_id(
     context, provider_id: str, description: str
 ) -> str | None:
@@ -229,6 +265,9 @@ async def call_provider_with_retry(
     retries = config_manager.get_llm_retries()
     backoff = config_manager.get_llm_backoff()
 
+    # 检查流式调用配置
+    enable_streaming_llm_call = config_manager.get_enable_streaming_llm_call()
+
     last_exc = None
     for attempt in range(1, retries + 1):
         try:
@@ -285,10 +324,16 @@ async def call_provider_with_retry(
                     if extra_generate_kwargs:
                         llm_kwargs.update(extra_generate_kwargs)
 
+                    if enable_streaming_llm_call:
+                        logger.info("[LLM 调用] 使用流式 Provider 调用")
+
+                    async def _invoke_llm(pid: str):
+                        if enable_streaming_llm_call:
+                            return await _call_provider_stream(context, pid, llm_kwargs)
+                        return await context.llm_generate(**llm_kwargs)
+
                     try:
-                        llm_resp = await context.llm_generate(
-                            **llm_kwargs,
-                        )
+                        llm_resp = await _invoke_llm(provider_id)
                     except Exception as e:
                         if (
                             response_format is not None
@@ -299,9 +344,7 @@ async def call_provider_with_retry(
                                 "已自动降级为无 schema 约束重试本次请求。"
                             )
                             llm_kwargs.pop("response_format", None)
-                            llm_resp = await context.llm_generate(
-                                **llm_kwargs,
-                            )
+                            llm_resp = await _invoke_llm(provider_id)
                         else:
                             raise
 
diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py
index fbaeac3..5f65382 100644
--- a/src/infrastructure/config/config_manager.py
+++ b/src/infrastructure/config/config_manager.py
@@ -194,6 +194,10 @@ class ConfigManager:
         """获取LLM请求重试退避基值（秒），实际退避会乘以尝试次数"""
         return self._get_group("llm").get("llm_backoff", 2)
 
+    def get_enable_streaming_llm_call(self) -> bool:
+        """获取是否启用流式 LLM 调用"""
+        return self._get_group("llm").get("enable_streaming_llm_call", False)
+
     def get_debug_mode(self) -> bool:
         """获取是否启用调试模式"""
         return self._get_group("basic").get("debug_mode", False)