fix(sse): correct UTF-8 handling and chunking in SSE streaming\n\n- Updated splitContentIntoSegments to handle runes instead of bytes\n- Fixed buildWebUIStreamForwarder to send delta chunks instead of cumulative content\n- Ensures proper handling of multi-byte characters in SSE streams
This commit is contained in:
@@ -27,6 +27,7 @@ Now supports mutually exclusive message channels:
|
|||||||
- If `telegram`: set `TELEGRAM_BOT_TOKEN`, keep `FEISHU_*` empty.
|
- If `telegram`: set `TELEGRAM_BOT_TOKEN`, keep `FEISHU_*` empty.
|
||||||
- If `feishu`: set `FEISHU_APP_ID` and `FEISHU_APP_SECRET`, keep `TELEGRAM_BOT_TOKEN` empty.
|
- If `feishu`: set `FEISHU_APP_ID` and `FEISHU_APP_SECRET`, keep `TELEGRAM_BOT_TOKEN` empty.
|
||||||
- If `webui`: set `WEBUI_LISTEN_ADDR` (default `:8090`).
|
- If `webui`: set `WEBUI_LISTEN_ADDR` (default `:8090`).
|
||||||
|
- Optional for `webui`: set `WEBUI_EXPOSE_REASONING=true` to expose `thought/tool_call/tool_result` events to frontend.
|
||||||
3. Set log level with `LOG_LEVEL=debug|info|warn|error`.
|
3. Set log level with `LOG_LEVEL=debug|info|warn|error`.
|
||||||
- To inspect full skill/tool execution content and detailed ReAct step traces, use `LOG_LEVEL=debug`.
|
- To inspect full skill/tool execution content and detailed ReAct step traces, use `LOG_LEVEL=debug`.
|
||||||
4. Configure knowledge and reasoning:
|
4. Configure knowledge and reasoning:
|
||||||
@@ -95,6 +96,7 @@ Minimum env config in `workspace/agent_runtime/configs/env`:
|
|||||||
```env
|
```env
|
||||||
MESSAGE_CHANNEL=webui
|
MESSAGE_CHANNEL=webui
|
||||||
WEBUI_LISTEN_ADDR=:8090
|
WEBUI_LISTEN_ADDR=:8090
|
||||||
|
WEBUI_EXPOSE_REASONING=false
|
||||||
LLM_API_KEY=your_api_key
|
LLM_API_KEY=your_api_key
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -115,6 +117,7 @@ docker run --rm -d \
|
|||||||
--env-file ./workspace/agent_runtime/configs/env \
|
--env-file ./workspace/agent_runtime/configs/env \
|
||||||
-e MESSAGE_CHANNEL=webui \
|
-e MESSAGE_CHANNEL=webui \
|
||||||
-e WEBUI_LISTEN_ADDR=:8090 \
|
-e WEBUI_LISTEN_ADDR=:8090 \
|
||||||
|
-e WEBUI_EXPOSE_REASONING=false \
|
||||||
-e AGENT_WORKSPACE_DIR=/app/workspace/agent_runtime \
|
-e AGENT_WORKSPACE_DIR=/app/workspace/agent_runtime \
|
||||||
-v "$(pwd)/workspace/agent_runtime:/app/workspace/agent_runtime" \
|
-v "$(pwd)/workspace/agent_runtime:/app/workspace/agent_runtime" \
|
||||||
laodingbot:latest
|
laodingbot:latest
|
||||||
|
|||||||
@@ -220,7 +220,7 @@ func runMessageChannel(ctx context.Context, cfg config.Config, engine *agent.Orc
|
|||||||
return engine.HandleMessage(ctx, msg.ChatID, msg.UserID, msg.Text)
|
return engine.HandleMessage(ctx, msg.ChatID, msg.UserID, msg.Text)
|
||||||
},
|
},
|
||||||
func(ctx context.Context, msg webui.IncomingMessage, callback webui.StreamEventCallback) (string, error) {
|
func(ctx context.Context, msg webui.IncomingMessage, callback webui.StreamEventCallback) (string, error) {
|
||||||
return engine.HandleMessageStream(ctx, msg.ChatID, msg.UserID, msg.Text, buildWebUIStreamForwarder(callback))
|
return engine.HandleMessageStream(ctx, msg.ChatID, msg.UserID, msg.Text, buildWebUIStreamForwarder(callback, cfg.WebUI.ExposeReasoning))
|
||||||
},
|
},
|
||||||
func(ctx context.Context, chatID, userID string, files []llm.InputFile) ([]string, error) {
|
func(ctx context.Context, chatID, userID string, files []llm.InputFile) ([]string, error) {
|
||||||
return engine.UploadAndCacheFiles(ctx, chatID, userID, files)
|
return engine.UploadAndCacheFiles(ctx, chatID, userID, files)
|
||||||
@@ -231,16 +231,54 @@ func runMessageChannel(ctx context.Context, cfg config.Config, engine *agent.Orc
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func buildWebUIStreamForwarder(callback webui.StreamEventCallback) agent.StreamEventCallback {
|
func buildWebUIStreamForwarder(callback webui.StreamEventCallback, exposeReasoning bool) agent.StreamEventCallback {
|
||||||
|
const finalChunkRunes = 12
|
||||||
|
const finalChunkInterval = 25 * time.Millisecond
|
||||||
|
|
||||||
return func(event agent.StreamEvent) error {
|
return func(event agent.StreamEvent) error {
|
||||||
if callback == nil {
|
if callback == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
switch event.Type {
|
switch event.Type {
|
||||||
case agent.StreamEventTypeFinal, agent.StreamEventTypeError:
|
case agent.StreamEventTypeThought, agent.StreamEventTypeToolCall, agent.StreamEventTypeToolResult:
|
||||||
|
if !exposeReasoning {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
return callback(webui.StreamEvent{
|
return callback(webui.StreamEvent{
|
||||||
Type: webui.StreamEventType(event.Type),
|
Type: webui.StreamEventType(event.Type),
|
||||||
Content: event.Content,
|
Content: event.Content,
|
||||||
|
Step: event.Step,
|
||||||
|
ToolName: event.ToolName,
|
||||||
|
})
|
||||||
|
case agent.StreamEventTypeFinal:
|
||||||
|
runes := []rune(event.Content)
|
||||||
|
if len(runes) == 0 {
|
||||||
|
return callback(webui.StreamEvent{Type: webui.StreamEventTypeFinal, Content: "", Step: event.Step})
|
||||||
|
}
|
||||||
|
start := 0
|
||||||
|
for start < len(runes) {
|
||||||
|
end := start + finalChunkRunes
|
||||||
|
if end > len(runes) {
|
||||||
|
end = len(runes)
|
||||||
|
}
|
||||||
|
if err := callback(webui.StreamEvent{
|
||||||
|
Type: webui.StreamEventTypeFinal,
|
||||||
|
Content: string(runes[start:end]),
|
||||||
|
Step: event.Step,
|
||||||
|
}); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
start = end
|
||||||
|
if start < len(runes) {
|
||||||
|
time.Sleep(finalChunkInterval)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
case agent.StreamEventTypeError:
|
||||||
|
return callback(webui.StreamEvent{
|
||||||
|
Type: webui.StreamEventTypeError,
|
||||||
|
Content: event.Content,
|
||||||
|
Step: event.Step,
|
||||||
})
|
})
|
||||||
default:
|
default:
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ FEISHU_LISTEN_ADDR=:8080
|
|||||||
FEISHU_EVENT_PATH=/feishu/events
|
FEISHU_EVENT_PATH=/feishu/events
|
||||||
WEBUI_LISTEN_ADDR=:8090
|
WEBUI_LISTEN_ADDR=:8090
|
||||||
WEBUI_MAX_UPLOAD_MB=20
|
WEBUI_MAX_UPLOAD_MB=20
|
||||||
|
WEBUI_EXPOSE_REASONING=false
|
||||||
|
|
||||||
LLM_BASE_URL=https://api.openai.com/v1
|
LLM_BASE_URL=https://api.openai.com/v1
|
||||||
LLM_API_KEY=
|
LLM_API_KEY=
|
||||||
|
|||||||
@@ -57,13 +57,17 @@ data: {"type":"final","content":"当前目录是 C:/Project/MyProject","step":2}
|
|||||||
|
|
||||||
事件类型:
|
事件类型:
|
||||||
|
|
||||||
|
- `thought`: LLM 思考片段(可选透传)
|
||||||
|
- `tool_call`: 工具调用请求(可选透传)
|
||||||
|
- `tool_result`: 工具执行结果(可选透传)
|
||||||
- `final`: 最终回答
|
- `final`: 最终回答
|
||||||
- `error`: 错误信息
|
- `error`: 错误信息
|
||||||
|
|
||||||
说明:
|
说明:
|
||||||
|
|
||||||
- 当前 WebUI 默认只向前端返回 `final` 和 `error`。
|
- 默认情况下(`WEBUI_EXPOSE_REASONING=false`),WebUI 只向前端返回 `final` 和 `error`。
|
||||||
- 中间推理轨迹(如 `thought`、`tool_call`、`tool_result` 以及对应的 `step`)会写入服务端 `debug` 日志,不再直接返回给用户界面。
|
- 当设置 `WEBUI_EXPOSE_REASONING=true` 时,WebUI 会额外透传 `thought`、`tool_call`、`tool_result` 事件。
|
||||||
|
- 无论是否透传推理事件,`final` 都会分段累计推送,以便前端实现打字机效果。
|
||||||
|
|
||||||
## 4. 连接生命周期
|
## 4. 连接生命周期
|
||||||
|
|
||||||
@@ -78,6 +82,11 @@ data: {"type":"final","content":"当前目录是 C:/Project/MyProject","step":2}
|
|||||||
- 答案区: 显示最后一个 `final`
|
- 答案区: 显示最后一个 `final`
|
||||||
- 错误区: 显示 `error`
|
- 错误区: 显示 `error`
|
||||||
|
|
||||||
|
如果开启了 `WEBUI_EXPOSE_REASONING=true`,建议额外提供“思考面板”:
|
||||||
|
|
||||||
|
- 思考区: 渲染 `thought`
|
||||||
|
- 工具区: 渲染 `tool_call` / `tool_result`
|
||||||
|
|
||||||
建议状态机:
|
建议状态机:
|
||||||
|
|
||||||
- `idle`: 初始状态
|
- `idle`: 初始状态
|
||||||
@@ -88,7 +97,7 @@ data: {"type":"final","content":"当前目录是 C:/Project/MyProject","step":2}
|
|||||||
## 6. TypeScript 对接示例 (fetch + ReadableStream)
|
## 6. TypeScript 对接示例 (fetch + ReadableStream)
|
||||||
|
|
||||||
```ts
|
```ts
|
||||||
type StreamEventType = 'final' | 'error';
|
type StreamEventType = 'thought' | 'tool_call' | 'tool_result' | 'final' | 'error';
|
||||||
|
|
||||||
interface StreamEvent {
|
interface StreamEvent {
|
||||||
type: StreamEventType;
|
type: StreamEventType;
|
||||||
@@ -165,7 +174,7 @@ export async function streamChat(
|
|||||||
- step?: number
|
- step?: number
|
||||||
- tool_name?: string
|
- tool_name?: string
|
||||||
5) 收到 final 视为本轮完成;收到 error 视为失败
|
5) 收到 final 视为本轮完成;收到 error 视为失败
|
||||||
6) 不要再假设前端会收到 thought/tool_call/tool_result;这些内部轨迹已改为服务端 debug 日志
|
6) 默认不要假设前端一定会收到 thought/tool_call/tool_result;仅当 `WEBUI_EXPOSE_REASONING=true` 才会透传
|
||||||
|
|
||||||
你的改造要求:
|
你的改造要求:
|
||||||
1) 保留现有 UI 风格和组件结构,不做无关重构
|
1) 保留现有 UI 风格和组件结构,不做无关重构
|
||||||
|
|||||||
@@ -731,14 +731,18 @@ func (o *Orchestrator) runUnifiedReActStream(ctx context.Context, chatID, userID
|
|||||||
|
|
||||||
// 推送思考过程事件
|
// 推送思考过程事件
|
||||||
if completion.Content != "" {
|
if completion.Content != "" {
|
||||||
|
// 分割内容为逐步推送的片段
|
||||||
|
segments := splitContentIntoSegments(completion.Content, 50) // 每段50字符
|
||||||
|
for _, segment := range segments {
|
||||||
if err := callback(StreamEvent{
|
if err := callback(StreamEvent{
|
||||||
Type: StreamEventTypeThought,
|
Type: StreamEventTypeThought,
|
||||||
Content: completion.Content,
|
Content: segment,
|
||||||
Step: step,
|
Step: step,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
return "", fmt.Errorf("callback error: %w", err)
|
return "", fmt.Errorf("callback error: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ========== 无 tool_calls → 最终回答 ==========
|
// ========== 无 tool_calls → 最终回答 ==========
|
||||||
if len(completion.ToolCalls) == 0 {
|
if len(completion.ToolCalls) == 0 {
|
||||||
@@ -1568,3 +1572,17 @@ func sanitizeUserFacingAnswer(raw string) string {
|
|||||||
}
|
}
|
||||||
return strings.TrimSpace(strings.Join(cleaned, "\n"))
|
return strings.TrimSpace(strings.Join(cleaned, "\n"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// splitContentIntoSegments splits a string into smaller segments of the specified size (by rune count).
|
||||||
|
func splitContentIntoSegments(content string, segmentSize int) []string {
|
||||||
|
runes := []rune(content)
|
||||||
|
var segments []string
|
||||||
|
for start := 0; start < len(runes); start += segmentSize {
|
||||||
|
end := start + segmentSize
|
||||||
|
if end > len(runes) {
|
||||||
|
end = len(runes)
|
||||||
|
}
|
||||||
|
segments = append(segments, string(runes[start:end]))
|
||||||
|
}
|
||||||
|
return segments
|
||||||
|
}
|
||||||
|
|||||||
@@ -50,6 +50,7 @@ type FeishuConfig struct {
|
|||||||
type WebUIConfig struct {
|
type WebUIConfig struct {
|
||||||
ListenAddr string
|
ListenAddr string
|
||||||
MaxUploadBytes int64
|
MaxUploadBytes int64
|
||||||
|
ExposeReasoning bool
|
||||||
}
|
}
|
||||||
|
|
||||||
type LLMConfig struct {
|
type LLMConfig struct {
|
||||||
@@ -112,6 +113,7 @@ func Load() (Config, error) {
|
|||||||
WebUI: WebUIConfig{
|
WebUI: WebUIConfig{
|
||||||
ListenAddr: defaultIfEmpty(os.Getenv("WEBUI_LISTEN_ADDR"), ":8090"),
|
ListenAddr: defaultIfEmpty(os.Getenv("WEBUI_LISTEN_ADDR"), ":8090"),
|
||||||
MaxUploadBytes: int64(intFromEnv("WEBUI_MAX_UPLOAD_MB", 20)) * 1024 * 1024,
|
MaxUploadBytes: int64(intFromEnv("WEBUI_MAX_UPLOAD_MB", 20)) * 1024 * 1024,
|
||||||
|
ExposeReasoning: boolFromEnv("WEBUI_EXPOSE_REASONING", false),
|
||||||
},
|
},
|
||||||
LLM: LLMConfig{
|
LLM: LLMConfig{
|
||||||
BaseURL: strings.TrimRight(defaultIfEmpty(os.Getenv("LLM_BASE_URL"), "https://api.openai.com/v1"), "/"),
|
BaseURL: strings.TrimRight(defaultIfEmpty(os.Getenv("LLM_BASE_URL"), "https://api.openai.com/v1"), "/"),
|
||||||
|
|||||||
Reference in New Issue
Block a user