Add streaming chat rendering for pico
This commit is contained in:
parent
6e1fab80e2
commit
d380f0b0ba
15 changed files with 568 additions and 16 deletions
244
pkg/agent/streaming_test.go
Normal file
244
pkg/agent/streaming_test.go
Normal file
|
|
@ -0,0 +1,244 @@
|
||||||
|
package agent
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"os"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/sipeed/picoclaw/pkg/bus"
|
||||||
|
"github.com/sipeed/picoclaw/pkg/config"
|
||||||
|
"github.com/sipeed/picoclaw/pkg/providers"
|
||||||
|
)
|
||||||
|
|
||||||
|
type streamingTestProvider struct {
|
||||||
|
chatCalled bool
|
||||||
|
chatStreamCalled bool
|
||||||
|
chunks []string
|
||||||
|
response *providers.LLMResponse
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *streamingTestProvider) Chat(
|
||||||
|
ctx context.Context,
|
||||||
|
messages []providers.Message,
|
||||||
|
tools []providers.ToolDefinition,
|
||||||
|
model string,
|
||||||
|
options map[string]any,
|
||||||
|
) (*providers.LLMResponse, error) {
|
||||||
|
p.chatCalled = true
|
||||||
|
if p.response != nil {
|
||||||
|
return p.response, nil
|
||||||
|
}
|
||||||
|
return &providers.LLMResponse{Content: "fallback"}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *streamingTestProvider) ChatStream(
|
||||||
|
ctx context.Context,
|
||||||
|
messages []providers.Message,
|
||||||
|
tools []providers.ToolDefinition,
|
||||||
|
model string,
|
||||||
|
options map[string]any,
|
||||||
|
onChunk func(accumulated string),
|
||||||
|
) (*providers.LLMResponse, error) {
|
||||||
|
p.chatStreamCalled = true
|
||||||
|
for _, chunk := range p.chunks {
|
||||||
|
onChunk(chunk)
|
||||||
|
}
|
||||||
|
if p.response != nil {
|
||||||
|
return p.response, nil
|
||||||
|
}
|
||||||
|
return &providers.LLMResponse{Content: "streamed"}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *streamingTestProvider) GetDefaultModel() string {
|
||||||
|
return "test-model"
|
||||||
|
}
|
||||||
|
|
||||||
|
type recordingStreamer struct {
|
||||||
|
updates []string
|
||||||
|
finalized []string
|
||||||
|
canceled bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *recordingStreamer) Update(ctx context.Context, content string) error {
|
||||||
|
s.updates = append(s.updates, content)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *recordingStreamer) Finalize(ctx context.Context, content string) error {
|
||||||
|
s.finalized = append(s.finalized, content)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *recordingStreamer) Cancel(ctx context.Context) {
|
||||||
|
s.canceled = true
|
||||||
|
}
|
||||||
|
|
||||||
|
type recordingStreamDelegate struct {
|
||||||
|
streamer *recordingStreamer
|
||||||
|
ok bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *recordingStreamDelegate) GetStreamer(
|
||||||
|
ctx context.Context,
|
||||||
|
channel, chatID string,
|
||||||
|
) (bus.Streamer, bool) {
|
||||||
|
if !d.ok {
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
|
return d.streamer, true
|
||||||
|
}
|
||||||
|
|
||||||
|
func newStreamingTestLoop(
|
||||||
|
t *testing.T,
|
||||||
|
streamingEnabled bool,
|
||||||
|
provider *streamingTestProvider,
|
||||||
|
streamer *recordingStreamer,
|
||||||
|
) (*AgentLoop, func()) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
tmpDir, err := os.MkdirTemp("", "agent-streaming-test-*")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("MkdirTemp() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg := &config.Config{
|
||||||
|
Agents: config.AgentsConfig{
|
||||||
|
Defaults: config.AgentDefaults{
|
||||||
|
Workspace: tmpDir,
|
||||||
|
ModelName: "test-model",
|
||||||
|
MaxTokens: 4096,
|
||||||
|
MaxToolIterations: 10,
|
||||||
|
StreamingEnabled: streamingEnabled,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
msgBus := bus.NewMessageBus()
|
||||||
|
msgBus.SetStreamDelegate(&recordingStreamDelegate{
|
||||||
|
streamer: streamer,
|
||||||
|
ok: true,
|
||||||
|
})
|
||||||
|
|
||||||
|
return NewAgentLoop(cfg, msgBus, provider), func() {
|
||||||
|
os.RemoveAll(tmpDir)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestProcessMessage_UsesStreamingProviderWhenEnabled(t *testing.T) {
|
||||||
|
provider := &streamingTestProvider{
|
||||||
|
chunks: []string{"Hel", "Hello"},
|
||||||
|
response: &providers.LLMResponse{
|
||||||
|
Content: "Hello",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
streamer := &recordingStreamer{}
|
||||||
|
al, cleanup := newStreamingTestLoop(t, true, provider, streamer)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
response, err := al.processMessage(context.Background(), testInboundMessage(bus.InboundMessage{
|
||||||
|
Channel: "pico",
|
||||||
|
ChatID: "pico:test-session",
|
||||||
|
SenderID: "user-1",
|
||||||
|
Content: "hello",
|
||||||
|
}))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("processMessage() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if response != "Hello" {
|
||||||
|
t.Fatalf("response = %q, want %q", response, "Hello")
|
||||||
|
}
|
||||||
|
if !provider.chatStreamCalled {
|
||||||
|
t.Fatal("expected ChatStream to be used")
|
||||||
|
}
|
||||||
|
if provider.chatCalled {
|
||||||
|
t.Fatal("expected Chat() not to be used when streaming is enabled")
|
||||||
|
}
|
||||||
|
if len(streamer.updates) != 2 {
|
||||||
|
t.Fatalf("stream updates = %#v, want 2 accumulated updates", streamer.updates)
|
||||||
|
}
|
||||||
|
if len(streamer.finalized) != 1 || streamer.finalized[0] != "Hello" {
|
||||||
|
t.Fatalf("finalized = %#v, want final Hello", streamer.finalized)
|
||||||
|
}
|
||||||
|
if streamer.canceled {
|
||||||
|
t.Fatal("streamer should not be canceled on direct response")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestProcessMessage_SkipsStreamingProviderWhenDisabled(t *testing.T) {
|
||||||
|
provider := &streamingTestProvider{
|
||||||
|
response: &providers.LLMResponse{
|
||||||
|
Content: "No stream",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
streamer := &recordingStreamer{}
|
||||||
|
al, cleanup := newStreamingTestLoop(t, false, provider, streamer)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
response, err := al.processMessage(context.Background(), testInboundMessage(bus.InboundMessage{
|
||||||
|
Channel: "pico",
|
||||||
|
ChatID: "pico:test-session",
|
||||||
|
SenderID: "user-1",
|
||||||
|
Content: "hello",
|
||||||
|
}))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("processMessage() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if response != "No stream" {
|
||||||
|
t.Fatalf("response = %q, want %q", response, "No stream")
|
||||||
|
}
|
||||||
|
if provider.chatStreamCalled {
|
||||||
|
t.Fatal("expected ChatStream not to be used when streaming is disabled")
|
||||||
|
}
|
||||||
|
if !provider.chatCalled {
|
||||||
|
t.Fatal("expected Chat() to be used when streaming is disabled")
|
||||||
|
}
|
||||||
|
if len(streamer.updates) != 0 || len(streamer.finalized) != 0 || streamer.canceled {
|
||||||
|
t.Fatalf("streamer should stay unused, got updates=%#v finalized=%#v canceled=%v", streamer.updates, streamer.finalized, streamer.canceled)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestProcessMessage_CancelsStreamingWhenToolCallsFollow(t *testing.T) {
|
||||||
|
provider := &streamingTestProvider{
|
||||||
|
chunks: []string{"Thinking"},
|
||||||
|
response: &providers.LLMResponse{
|
||||||
|
Content: "Thinking",
|
||||||
|
ToolCalls: []providers.ToolCall{
|
||||||
|
{
|
||||||
|
ID: "call-1",
|
||||||
|
Name: "missing_tool",
|
||||||
|
Arguments: map[string]any{
|
||||||
|
"q": "stream",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
streamer := &recordingStreamer{}
|
||||||
|
al, cleanup := newStreamingTestLoop(t, true, provider, streamer)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
response, err := al.processMessage(context.Background(), testInboundMessage(bus.InboundMessage{
|
||||||
|
Channel: "pico",
|
||||||
|
ChatID: "pico:test-session",
|
||||||
|
SenderID: "user-1",
|
||||||
|
Content: "hello",
|
||||||
|
}))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("processMessage() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if response == "" {
|
||||||
|
t.Fatal("expected follow-up tool result content after tool-call path")
|
||||||
|
}
|
||||||
|
if !provider.chatStreamCalled {
|
||||||
|
t.Fatal("expected ChatStream to be used")
|
||||||
|
}
|
||||||
|
if !streamer.canceled {
|
||||||
|
t.Fatal("expected streamer to be canceled when tool calls follow streamed text")
|
||||||
|
}
|
||||||
|
if len(streamer.finalized) != 0 {
|
||||||
|
t.Fatalf("finalized = %#v, want no finalize after tool calls", streamer.finalized)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -107,6 +107,16 @@ type PicoChannel struct {
|
||||||
deleteMessageFn func(context.Context, string, string) error
|
deleteMessageFn func(context.Context, string, string) error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type picoStreamer struct {
|
||||||
|
channel *PicoChannel
|
||||||
|
chatID string
|
||||||
|
messageID string
|
||||||
|
|
||||||
|
mu sync.Mutex
|
||||||
|
started bool
|
||||||
|
closed bool
|
||||||
|
}
|
||||||
|
|
||||||
// NewPicoChannel creates a new Pico Protocol channel.
|
// NewPicoChannel creates a new Pico Protocol channel.
|
||||||
func NewPicoChannel(
|
func NewPicoChannel(
|
||||||
bc *config.Channel,
|
bc *config.Channel,
|
||||||
|
|
@ -354,6 +364,18 @@ func (c *PicoChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]stri
|
||||||
return []string{msgID}, nil
|
return []string{msgID}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (c *PicoChannel) BeginStream(_ context.Context, chatID string) (channels.Streamer, error) {
|
||||||
|
if !c.IsRunning() {
|
||||||
|
return nil, channels.ErrNotRunning
|
||||||
|
}
|
||||||
|
|
||||||
|
return &picoStreamer{
|
||||||
|
channel: c,
|
||||||
|
chatID: chatID,
|
||||||
|
messageID: uuid.New().String(),
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
// EditMessage implements channels.MessageEditor.
|
// EditMessage implements channels.MessageEditor.
|
||||||
func (c *PicoChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error {
|
func (c *PicoChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error {
|
||||||
return c.editMessage(ctx, chatID, messageID, content, nil)
|
return c.editMessage(ctx, chatID, messageID, content, nil)
|
||||||
|
|
@ -440,6 +462,74 @@ func (c *PicoChannel) FinalizeToolFeedbackMessage(ctx context.Context, msg bus.O
|
||||||
return c.finalizeTrackedToolFeedbackMessage(ctx, msg.ChatID, msg.Content, c.editMessage, msg.ContextUsage)
|
return c.finalizeTrackedToolFeedbackMessage(ctx, msg.ChatID, msg.Content, c.editMessage, msg.ContextUsage)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *picoStreamer) Update(_ context.Context, content string) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
if s.closed {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if !s.started {
|
||||||
|
s.started = true
|
||||||
|
return s.channel.broadcastToSession(s.chatID, newMessage(TypeMessageCreate, map[string]any{
|
||||||
|
"message_id": s.messageID,
|
||||||
|
PayloadKeyContent: content,
|
||||||
|
PayloadKeyThought: false,
|
||||||
|
PayloadKeyStreaming: true,
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.channel.broadcastToSession(s.chatID, newMessage(TypeMessageUpdate, map[string]any{
|
||||||
|
"message_id": s.messageID,
|
||||||
|
"content": content,
|
||||||
|
"streaming": true,
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *picoStreamer) Finalize(_ context.Context, content string) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
if s.closed {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
s.closed = true
|
||||||
|
|
||||||
|
if !s.started {
|
||||||
|
return s.channel.broadcastToSession(s.chatID, newMessage(TypeMessageCreate, map[string]any{
|
||||||
|
"message_id": s.messageID,
|
||||||
|
PayloadKeyContent: content,
|
||||||
|
PayloadKeyThought: false,
|
||||||
|
PayloadKeyStreaming: false,
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.channel.broadcastToSession(s.chatID, newMessage(TypeMessageUpdate, map[string]any{
|
||||||
|
"message_id": s.messageID,
|
||||||
|
"content": content,
|
||||||
|
"streaming": false,
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *picoStreamer) Cancel(_ context.Context) {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
if s.closed {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.closed = true
|
||||||
|
|
||||||
|
if !s.started {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
_ = s.channel.broadcastToSession(s.chatID, newMessage(TypeMessageDelete, map[string]any{
|
||||||
|
"message_id": s.messageID,
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
// StartTyping implements channels.TypingCapable.
|
// StartTyping implements channels.TypingCapable.
|
||||||
func (c *PicoChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
|
func (c *PicoChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
|
||||||
startMsg := newMessage(TypeTypingStart, nil)
|
startMsg := newMessage(TypeTypingStart, nil)
|
||||||
|
|
|
||||||
|
|
@ -22,10 +22,13 @@ const (
|
||||||
TypeError = "error"
|
TypeError = "error"
|
||||||
TypePong = "pong"
|
TypePong = "pong"
|
||||||
|
|
||||||
|
PicoTokenPrefix = "pico-"
|
||||||
|
|
||||||
PayloadKeyContent = "content"
|
PayloadKeyContent = "content"
|
||||||
PayloadKeyThought = "thought"
|
PayloadKeyThought = "thought"
|
||||||
PayloadKeyKind = "kind"
|
PayloadKeyKind = "kind"
|
||||||
PayloadKeyToolCalls = "tool_calls"
|
PayloadKeyToolCalls = "tool_calls"
|
||||||
|
PayloadKeyStreaming = "streaming"
|
||||||
|
|
||||||
MessageKindThought = "thought"
|
MessageKindThought = "thought"
|
||||||
MessageKindToolCalls = "tool_calls"
|
MessageKindToolCalls = "tool_calls"
|
||||||
|
|
|
||||||
|
|
@ -273,6 +273,7 @@ type AgentDefaults struct {
|
||||||
MaxParallelTurns int `json:"max_parallel_turns,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_PARALLEL_TURNS"` // Max concurrent turns (0 or 1 = sequential)
|
MaxParallelTurns int `json:"max_parallel_turns,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_PARALLEL_TURNS"` // Max concurrent turns (0 or 1 = sequential)
|
||||||
SubTurn SubTurnConfig `json:"subturn" envPrefix:"PICOCLAW_AGENTS_DEFAULTS_SUBTURN_"`
|
SubTurn SubTurnConfig `json:"subturn" envPrefix:"PICOCLAW_AGENTS_DEFAULTS_SUBTURN_"`
|
||||||
ToolFeedback ToolFeedbackConfig `json:"tool_feedback,omitempty"`
|
ToolFeedback ToolFeedbackConfig `json:"tool_feedback,omitempty"`
|
||||||
|
StreamingEnabled bool `json:"streaming_enabled" env:"PICOCLAW_AGENTS_DEFAULTS_STREAMING_ENABLED"`
|
||||||
SplitOnMarker bool `json:"split_on_marker" env:"PICOCLAW_AGENTS_DEFAULTS_SPLIT_ON_MARKER"` // split messages on <|[SPLIT]|> marker
|
SplitOnMarker bool `json:"split_on_marker" env:"PICOCLAW_AGENTS_DEFAULTS_SPLIT_ON_MARKER"` // split messages on <|[SPLIT]|> marker
|
||||||
ContextManager string `json:"context_manager,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER"`
|
ContextManager string `json:"context_manager,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER"`
|
||||||
ContextManagerConfig json.RawMessage `json:"context_manager_config,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER_CONFIG"`
|
ContextManagerConfig json.RawMessage `json:"context_manager_config,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER_CONFIG"`
|
||||||
|
|
|
||||||
|
|
@ -39,6 +39,7 @@ func DefaultConfig() *Config {
|
||||||
MaxArgsLength: 300,
|
MaxArgsLength: 300,
|
||||||
SeparateMessages: false,
|
SeparateMessages: false,
|
||||||
},
|
},
|
||||||
|
StreamingEnabled: true,
|
||||||
SplitOnMarker: false,
|
SplitOnMarker: false,
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -9,12 +9,8 @@ import {
|
||||||
} from "@tabler/icons-react"
|
} from "@tabler/icons-react"
|
||||||
import { useState } from "react"
|
import { useState } from "react"
|
||||||
import { useTranslation } from "react-i18next"
|
import { useTranslation } from "react-i18next"
|
||||||
import ReactMarkdown from "react-markdown"
|
|
||||||
import rehypeHighlight from "rehype-highlight"
|
|
||||||
import rehypeRaw from "rehype-raw"
|
|
||||||
import rehypeSanitize from "rehype-sanitize"
|
|
||||||
import remarkGfm from "remark-gfm"
|
|
||||||
|
|
||||||
|
import { StreamingMarkdown } from "@/components/chat/streaming-markdown"
|
||||||
import { Button } from "@/components/ui/button"
|
import { Button } from "@/components/ui/button"
|
||||||
import { formatMessageTime } from "@/hooks/use-pico-chat"
|
import { formatMessageTime } from "@/hooks/use-pico-chat"
|
||||||
import { cn } from "@/lib/utils"
|
import { cn } from "@/lib/utils"
|
||||||
|
|
@ -29,6 +25,8 @@ interface AssistantMessageProps {
|
||||||
attachments?: ChatAttachment[]
|
attachments?: ChatAttachment[]
|
||||||
kind?: AssistantMessageKind
|
kind?: AssistantMessageKind
|
||||||
toolCalls?: ChatToolCall[]
|
toolCalls?: ChatToolCall[]
|
||||||
|
streamingEnabled?: boolean
|
||||||
|
isStreaming?: boolean
|
||||||
timestamp?: string | number
|
timestamp?: string | number
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -37,6 +35,8 @@ export function AssistantMessage({
|
||||||
attachments = [],
|
attachments = [],
|
||||||
kind = "normal",
|
kind = "normal",
|
||||||
toolCalls = [],
|
toolCalls = [],
|
||||||
|
streamingEnabled = true,
|
||||||
|
isStreaming = false,
|
||||||
timestamp = "",
|
timestamp = "",
|
||||||
}: AssistantMessageProps) {
|
}: AssistantMessageProps) {
|
||||||
const { t } = useTranslation()
|
const { t } = useTranslation()
|
||||||
|
|
@ -184,21 +184,16 @@ export function AssistantMessage({
|
||||||
</div>
|
</div>
|
||||||
)}
|
)}
|
||||||
{(!isCollapsedBlock || isExpanded) && !isToolCalls && hasText && (
|
{(!isCollapsedBlock || isExpanded) && !isToolCalls && hasText && (
|
||||||
<div
|
<StreamingMarkdown
|
||||||
|
content={content}
|
||||||
|
animate={streamingEnabled && isStreaming}
|
||||||
className={cn(
|
className={cn(
|
||||||
"prose dark:prose-invert prose-pre:my-2 prose-pre:overflow-x-auto prose-pre:rounded-lg prose-pre:border prose-pre:bg-zinc-100 prose-pre:p-0 prose-pre:text-zinc-900 dark:prose-pre:bg-zinc-950 dark:prose-pre:text-zinc-100 max-w-none [overflow-wrap:anywhere] break-words",
|
"prose dark:prose-invert prose-pre:my-2 prose-pre:overflow-x-auto prose-pre:rounded-lg prose-pre:border prose-pre:bg-zinc-100 prose-pre:p-0 prose-pre:text-zinc-900 dark:prose-pre:bg-zinc-950 dark:prose-pre:text-zinc-100 max-w-none [overflow-wrap:anywhere] break-words",
|
||||||
isThought
|
isThought
|
||||||
? "prose-p:my-1.5 prose-p:whitespace-pre-wrap px-3 pt-0 pb-3 text-[13px] leading-relaxed opacity-70"
|
? "prose-p:my-1.5 prose-p:whitespace-pre-wrap px-3 pt-0 pb-3 text-[13px] leading-relaxed opacity-70"
|
||||||
: "prose-p:my-2 prose-p:whitespace-pre-wrap p-4 text-[15px] leading-relaxed",
|
: "prose-p:my-2 prose-p:whitespace-pre-wrap p-4 text-[15px] leading-relaxed",
|
||||||
)}
|
)}
|
||||||
>
|
/>
|
||||||
<ReactMarkdown
|
|
||||||
remarkPlugins={[remarkGfm]}
|
|
||||||
rehypePlugins={[rehypeRaw, rehypeSanitize, rehypeHighlight]}
|
|
||||||
>
|
|
||||||
{content}
|
|
||||||
</ReactMarkdown>
|
|
||||||
</div>
|
|
||||||
)}
|
)}
|
||||||
|
|
||||||
{!isCollapsedBlock && hasText && (
|
{!isCollapsedBlock && hasText && (
|
||||||
|
|
|
||||||
|
|
@ -4,6 +4,7 @@ import { type ChangeEvent, useEffect, useRef, useState } from "react"
|
||||||
import { useTranslation } from "react-i18next"
|
import { useTranslation } from "react-i18next"
|
||||||
import { toast } from "sonner"
|
import { toast } from "sonner"
|
||||||
|
|
||||||
|
import { launcherFetch } from "@/api/http"
|
||||||
import { AssistantMessage } from "@/components/chat/assistant-message"
|
import { AssistantMessage } from "@/components/chat/assistant-message"
|
||||||
import {
|
import {
|
||||||
ChatComposer,
|
ChatComposer,
|
||||||
|
|
@ -21,6 +22,7 @@ import { useChatModels } from "@/hooks/use-chat-models"
|
||||||
import { useGateway } from "@/hooks/use-gateway"
|
import { useGateway } from "@/hooks/use-gateway"
|
||||||
import { usePicoChat } from "@/hooks/use-pico-chat"
|
import { usePicoChat } from "@/hooks/use-pico-chat"
|
||||||
import { useSessionHistory } from "@/hooks/use-session-history"
|
import { useSessionHistory } from "@/hooks/use-session-history"
|
||||||
|
import { useQuery } from "@tanstack/react-query"
|
||||||
import type { ConnectionState } from "@/store/chat"
|
import type { ConnectionState } from "@/store/chat"
|
||||||
import type { ChatAttachment } from "@/store/chat"
|
import type { ChatAttachment } from "@/store/chat"
|
||||||
import { showAssistantDetailsAtom } from "@/store/chat"
|
import { showAssistantDetailsAtom } from "@/store/chat"
|
||||||
|
|
@ -159,6 +161,23 @@ export function ChatPage() {
|
||||||
onDeletedActiveSession: newChat,
|
onDeletedActiveSession: newChat,
|
||||||
})
|
})
|
||||||
|
|
||||||
|
const { data: appConfig } = useQuery({
|
||||||
|
queryKey: ["config"],
|
||||||
|
queryFn: async () => {
|
||||||
|
const res = await launcherFetch("/api/config")
|
||||||
|
if (!res.ok) {
|
||||||
|
throw new Error("Failed to load config")
|
||||||
|
}
|
||||||
|
return res.json() as Promise<Record<string, unknown>>
|
||||||
|
},
|
||||||
|
staleTime: 60_000,
|
||||||
|
})
|
||||||
|
const streamingEnabled =
|
||||||
|
(
|
||||||
|
(appConfig?.agents as { defaults?: { streaming_enabled?: boolean } })
|
||||||
|
?.defaults?.streaming_enabled
|
||||||
|
) !== false
|
||||||
|
|
||||||
const syncScrollState = (element: HTMLDivElement) => {
|
const syncScrollState = (element: HTMLDivElement) => {
|
||||||
const { clientHeight, scrollHeight, scrollTop } = element
|
const { clientHeight, scrollHeight, scrollTop } = element
|
||||||
setHasScrolled(scrollTop > 0)
|
setHasScrolled(scrollTop > 0)
|
||||||
|
|
@ -340,6 +359,8 @@ export function ChatPage() {
|
||||||
attachments={msg.attachments}
|
attachments={msg.attachments}
|
||||||
kind={msg.kind}
|
kind={msg.kind}
|
||||||
toolCalls={msg.toolCalls}
|
toolCalls={msg.toolCalls}
|
||||||
|
isStreaming={msg.streaming}
|
||||||
|
streamingEnabled={streamingEnabled}
|
||||||
timestamp={msg.timestamp}
|
timestamp={msg.timestamp}
|
||||||
/>
|
/>
|
||||||
) : (
|
) : (
|
||||||
|
|
|
||||||
109
web/frontend/src/components/chat/streaming-markdown.tsx
Normal file
109
web/frontend/src/components/chat/streaming-markdown.tsx
Normal file
|
|
@ -0,0 +1,109 @@
|
||||||
|
import { useEffect, useMemo, useState } from "react"
|
||||||
|
import ReactMarkdown from "react-markdown"
|
||||||
|
import rehypeHighlight from "rehype-highlight"
|
||||||
|
import rehypeRaw from "rehype-raw"
|
||||||
|
import rehypeSanitize from "rehype-sanitize"
|
||||||
|
import remarkGfm from "remark-gfm"
|
||||||
|
|
||||||
|
import { cn } from "@/lib/utils"
|
||||||
|
|
||||||
|
interface StreamingMarkdownProps {
|
||||||
|
content: string
|
||||||
|
animate: boolean
|
||||||
|
className?: string
|
||||||
|
}
|
||||||
|
|
||||||
|
function nextChunkSize(remaining: number) {
|
||||||
|
if (remaining > 480) return 6
|
||||||
|
if (remaining > 240) return 4
|
||||||
|
if (remaining > 120) return 3
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
|
||||||
|
function nextFrameDelay(nextChunk: string, remaining: number) {
|
||||||
|
if (/[,.!?;:\n]$/.test(nextChunk)) {
|
||||||
|
return 90
|
||||||
|
}
|
||||||
|
if (remaining > 480) return 32
|
||||||
|
if (remaining > 240) return 38
|
||||||
|
if (remaining > 120) return 44
|
||||||
|
return 52
|
||||||
|
}
|
||||||
|
|
||||||
|
export function StreamingMarkdown({
|
||||||
|
content,
|
||||||
|
animate,
|
||||||
|
className,
|
||||||
|
}: StreamingMarkdownProps) {
|
||||||
|
const [displayedContent, setDisplayedContent] = useState(() =>
|
||||||
|
animate ? "" : content,
|
||||||
|
)
|
||||||
|
|
||||||
|
useEffect(() => {
|
||||||
|
if (!animate) {
|
||||||
|
setDisplayedContent(content)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
setDisplayedContent((current) => {
|
||||||
|
if (current === "" && content.length > 0) {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
if (content.length < current.length) {
|
||||||
|
return content
|
||||||
|
}
|
||||||
|
return current
|
||||||
|
})
|
||||||
|
}, [animate, content])
|
||||||
|
|
||||||
|
useEffect(() => {
|
||||||
|
if (!animate || displayedContent === content) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
const remaining = content.length - displayedContent.length
|
||||||
|
const chunkSize = nextChunkSize(remaining)
|
||||||
|
const nextChunk = content.slice(
|
||||||
|
displayedContent.length,
|
||||||
|
displayedContent.length + chunkSize,
|
||||||
|
)
|
||||||
|
|
||||||
|
const timer = window.setTimeout(() => {
|
||||||
|
setDisplayedContent((current) => {
|
||||||
|
if (current === content) {
|
||||||
|
return current
|
||||||
|
}
|
||||||
|
|
||||||
|
const nextLength = Math.min(
|
||||||
|
content.length,
|
||||||
|
current.length + chunkSize,
|
||||||
|
)
|
||||||
|
return content.slice(0, nextLength)
|
||||||
|
})
|
||||||
|
}, nextFrameDelay(nextChunk, remaining))
|
||||||
|
|
||||||
|
return () => window.clearTimeout(timer)
|
||||||
|
}, [animate, content, displayedContent])
|
||||||
|
|
||||||
|
const markdown = useMemo(
|
||||||
|
() => (animate ? displayedContent : content),
|
||||||
|
[animate, content, displayedContent],
|
||||||
|
)
|
||||||
|
|
||||||
|
return (
|
||||||
|
<div className={cn("relative", className)}>
|
||||||
|
<ReactMarkdown
|
||||||
|
remarkPlugins={[remarkGfm]}
|
||||||
|
rehypePlugins={[rehypeRaw, rehypeSanitize, rehypeHighlight]}
|
||||||
|
>
|
||||||
|
{markdown}
|
||||||
|
</ReactMarkdown>
|
||||||
|
{animate && (
|
||||||
|
<span
|
||||||
|
aria-hidden="true"
|
||||||
|
className="bg-foreground/70 ml-0.5 inline-block h-[1.05em] w-0.5 animate-pulse align-[-0.12em]"
|
||||||
|
/>
|
||||||
|
)}
|
||||||
|
</div>
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
@ -242,6 +242,7 @@ export function ConfigPage() {
|
||||||
defaults: {
|
defaults: {
|
||||||
workspace,
|
workspace,
|
||||||
restrict_to_workspace: form.restrictToWorkspace,
|
restrict_to_workspace: form.restrictToWorkspace,
|
||||||
|
streaming_enabled: form.streamingEnabled,
|
||||||
split_on_marker: form.splitOnMarker,
|
split_on_marker: form.splitOnMarker,
|
||||||
tool_feedback: {
|
tool_feedback: {
|
||||||
enabled: form.toolFeedbackEnabled,
|
enabled: form.toolFeedbackEnabled,
|
||||||
|
|
|
||||||
|
|
@ -95,6 +95,16 @@ export function AgentDefaultsSection({
|
||||||
}
|
}
|
||||||
/>
|
/>
|
||||||
|
|
||||||
|
<SwitchCardField
|
||||||
|
label={t("pages.config.streaming_enabled")}
|
||||||
|
hint={t("pages.config.streaming_enabled_hint")}
|
||||||
|
layout="setting-row"
|
||||||
|
checked={form.streamingEnabled}
|
||||||
|
onCheckedChange={(checked) =>
|
||||||
|
onFieldChange("streamingEnabled", checked)
|
||||||
|
}
|
||||||
|
/>
|
||||||
|
|
||||||
<SwitchCardField
|
<SwitchCardField
|
||||||
label={t("pages.config.split_on_marker")}
|
label={t("pages.config.split_on_marker")}
|
||||||
hint={t("pages.config.split_on_marker_hint")}
|
hint={t("pages.config.split_on_marker_hint")}
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,7 @@ export type JsonRecord = Record<string, unknown>
|
||||||
export interface CoreConfigForm {
|
export interface CoreConfigForm {
|
||||||
workspace: string
|
workspace: string
|
||||||
restrictToWorkspace: boolean
|
restrictToWorkspace: boolean
|
||||||
|
streamingEnabled: boolean
|
||||||
splitOnMarker: boolean
|
splitOnMarker: boolean
|
||||||
toolFeedbackEnabled: boolean
|
toolFeedbackEnabled: boolean
|
||||||
toolFeedbackMaxArgsLength: string
|
toolFeedbackMaxArgsLength: string
|
||||||
|
|
@ -69,6 +70,7 @@ export const DM_SCOPE_OPTIONS = [
|
||||||
export const EMPTY_FORM: CoreConfigForm = {
|
export const EMPTY_FORM: CoreConfigForm = {
|
||||||
workspace: "",
|
workspace: "",
|
||||||
restrictToWorkspace: true,
|
restrictToWorkspace: true,
|
||||||
|
streamingEnabled: true,
|
||||||
splitOnMarker: false,
|
splitOnMarker: false,
|
||||||
toolFeedbackEnabled: false,
|
toolFeedbackEnabled: false,
|
||||||
toolFeedbackMaxArgsLength: "300",
|
toolFeedbackMaxArgsLength: "300",
|
||||||
|
|
@ -144,6 +146,10 @@ export function buildFormFromConfig(config: unknown): CoreConfigForm {
|
||||||
defaults.restrict_to_workspace === undefined
|
defaults.restrict_to_workspace === undefined
|
||||||
? EMPTY_FORM.restrictToWorkspace
|
? EMPTY_FORM.restrictToWorkspace
|
||||||
: asBool(defaults.restrict_to_workspace),
|
: asBool(defaults.restrict_to_workspace),
|
||||||
|
streamingEnabled:
|
||||||
|
defaults.streaming_enabled === undefined
|
||||||
|
? EMPTY_FORM.streamingEnabled
|
||||||
|
: asBool(defaults.streaming_enabled),
|
||||||
splitOnMarker:
|
splitOnMarker:
|
||||||
defaults.split_on_marker === undefined
|
defaults.split_on_marker === undefined
|
||||||
? EMPTY_FORM.splitOnMarker
|
? EMPTY_FORM.splitOnMarker
|
||||||
|
|
|
||||||
|
|
@ -7,6 +7,7 @@ import {
|
||||||
import { normalizeUnixTimestamp } from "@/features/chat/state"
|
import { normalizeUnixTimestamp } from "@/features/chat/state"
|
||||||
import {
|
import {
|
||||||
type ChatAttachment,
|
type ChatAttachment,
|
||||||
|
type ChatMessage,
|
||||||
type ContextUsage,
|
type ContextUsage,
|
||||||
updateChatStore,
|
updateChatStore,
|
||||||
} from "@/store/chat"
|
} from "@/store/chat"
|
||||||
|
|
@ -83,6 +84,38 @@ function parseContextUsage(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function isToolFeedbackMessage(message: ChatMessage): boolean {
|
||||||
|
if (message.role !== "assistant") {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
const firstLine = message.content.split("\n", 1)[0]?.trim() ?? ""
|
||||||
|
return /^🔧\s+`[^`]+`/.test(firstLine)
|
||||||
|
}
|
||||||
|
|
||||||
|
function findToolFeedbackMessageIndex(messages: ChatMessage[]): number {
|
||||||
|
let lastUserIndex = -1
|
||||||
|
for (let i = messages.length - 1; i >= 0; i -= 1) {
|
||||||
|
if (messages[i].role === "user") {
|
||||||
|
lastUserIndex = i
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for (let i = messages.length - 1; i >= 0; i -= 1) {
|
||||||
|
if (i <= lastUserIndex) {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
if (isToolFeedbackMessage(messages[i])) {
|
||||||
|
return i
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return -1
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseStreamingFlag(payload: Record<string, unknown>): boolean | undefined {
|
||||||
|
return typeof payload.streaming === "boolean" ? payload.streaming : undefined
|
||||||
|
}
|
||||||
export function handlePicoMessage(
|
export function handlePicoMessage(
|
||||||
message: PicoMessage,
|
message: PicoMessage,
|
||||||
expectedSessionId: string,
|
expectedSessionId: string,
|
||||||
|
|
@ -101,6 +134,7 @@ export function handlePicoMessage(
|
||||||
parseAssistantMessageCreateState(payload)
|
parseAssistantMessageCreateState(payload)
|
||||||
const attachments = parseAttachments(payload)
|
const attachments = parseAttachments(payload)
|
||||||
const contextUsage = parseContextUsage(payload)
|
const contextUsage = parseContextUsage(payload)
|
||||||
|
const streaming = parseStreamingFlag(payload)
|
||||||
const timestamp =
|
const timestamp =
|
||||||
message.timestamp !== undefined &&
|
message.timestamp !== undefined &&
|
||||||
Number.isFinite(Number(message.timestamp))
|
Number.isFinite(Number(message.timestamp))
|
||||||
|
|
@ -117,6 +151,7 @@ export function handlePicoMessage(
|
||||||
kind,
|
kind,
|
||||||
...(toolCalls ? { toolCalls } : {}),
|
...(toolCalls ? { toolCalls } : {}),
|
||||||
attachments,
|
attachments,
|
||||||
|
...(streaming !== undefined ? { streaming } : {}),
|
||||||
timestamp,
|
timestamp,
|
||||||
},
|
},
|
||||||
],
|
],
|
||||||
|
|
@ -135,6 +170,7 @@ export function handlePicoMessage(
|
||||||
Number.isFinite(Number(message.timestamp))
|
Number.isFinite(Number(message.timestamp))
|
||||||
? normalizeUnixTimestamp(Number(message.timestamp))
|
? normalizeUnixTimestamp(Number(message.timestamp))
|
||||||
: Date.now()
|
: Date.now()
|
||||||
|
const streaming = parseStreamingFlag(payload)
|
||||||
if (!messageId) {
|
if (!messageId) {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
@ -156,6 +192,7 @@ export function handlePicoMessage(
|
||||||
kind,
|
kind,
|
||||||
toolCalls,
|
toolCalls,
|
||||||
...(attachments ? { attachments } : {}),
|
...(attachments ? { attachments } : {}),
|
||||||
|
...(streaming !== undefined ? { streaming } : {}),
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
if (found) {
|
if (found) {
|
||||||
|
|
@ -164,6 +201,22 @@ export function handlePicoMessage(
|
||||||
|
|
||||||
const { content, kind, toolCalls } =
|
const { content, kind, toolCalls } =
|
||||||
parseAssistantMessageUpdateState(payload)
|
parseAssistantMessageUpdateState(payload)
|
||||||
|
const fallbackIndex = findToolFeedbackMessageIndex(messages)
|
||||||
|
if (fallbackIndex >= 0) {
|
||||||
|
return messages.map((msg, index) =>
|
||||||
|
index === fallbackIndex
|
||||||
|
? {
|
||||||
|
...msg,
|
||||||
|
id: messageId,
|
||||||
|
content,
|
||||||
|
kind,
|
||||||
|
toolCalls,
|
||||||
|
...(attachments ? { attachments } : {}),
|
||||||
|
...(streaming !== undefined ? { streaming } : {}),
|
||||||
|
}
|
||||||
|
: msg,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
return [
|
return [
|
||||||
...messages,
|
...messages,
|
||||||
|
|
@ -174,6 +227,7 @@ export function handlePicoMessage(
|
||||||
kind,
|
kind,
|
||||||
toolCalls,
|
toolCalls,
|
||||||
...(attachments ? { attachments } : {}),
|
...(attachments ? { attachments } : {}),
|
||||||
|
...(streaming !== undefined ? { streaming } : {}),
|
||||||
timestamp,
|
timestamp,
|
||||||
},
|
},
|
||||||
]
|
]
|
||||||
|
|
@ -190,7 +244,19 @@ export function handlePicoMessage(
|
||||||
}
|
}
|
||||||
|
|
||||||
updateChatStore((prev) => ({
|
updateChatStore((prev) => ({
|
||||||
messages: prev.messages.filter((msg) => msg.id !== messageId),
|
messages: (() => {
|
||||||
|
const exactMessages = prev.messages.filter((msg) => msg.id !== messageId)
|
||||||
|
if (exactMessages.length !== prev.messages.length) {
|
||||||
|
return exactMessages
|
||||||
|
}
|
||||||
|
|
||||||
|
const fallbackIndex = findToolFeedbackMessageIndex(prev.messages)
|
||||||
|
if (fallbackIndex < 0) {
|
||||||
|
return prev.messages
|
||||||
|
}
|
||||||
|
|
||||||
|
return prev.messages.filter((_, index) => index !== fallbackIndex)
|
||||||
|
})(),
|
||||||
}))
|
}))
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -617,6 +617,8 @@
|
||||||
"workspace_hint": "Base directory for agent file operations.",
|
"workspace_hint": "Base directory for agent file operations.",
|
||||||
"restrict_workspace": "Restrict to Workspace",
|
"restrict_workspace": "Restrict to Workspace",
|
||||||
"restrict_workspace_hint": "Only allow file operations inside workspace.",
|
"restrict_workspace_hint": "Only allow file operations inside workspace.",
|
||||||
|
"streaming_enabled": "Streaming Response",
|
||||||
|
"streaming_enabled_hint": "Render live model output in chat and keep markdown updating as new text arrives.",
|
||||||
"split_on_marker": "Chatty Mode",
|
"split_on_marker": "Chatty Mode",
|
||||||
"split_on_marker_hint": "Split long messages into short ones like real human chatting.",
|
"split_on_marker_hint": "Split long messages into short ones like real human chatting.",
|
||||||
"tool_feedback_enabled": "Tool Feedback",
|
"tool_feedback_enabled": "Tool Feedback",
|
||||||
|
|
|
||||||
|
|
@ -617,6 +617,8 @@
|
||||||
"workspace_hint": "智能体执行文件读写操作时使用的基础目录",
|
"workspace_hint": "智能体执行文件读写操作时使用的基础目录",
|
||||||
"restrict_workspace": "限制工作目录访问",
|
"restrict_workspace": "限制工作目录访问",
|
||||||
"restrict_workspace_hint": "仅允许在工作目录内执行文件操作",
|
"restrict_workspace_hint": "仅允许在工作目录内执行文件操作",
|
||||||
|
"streaming_enabled": "流式响应",
|
||||||
|
"streaming_enabled_hint": "在对话中实时渲染模型输出,并在新文本到达时持续更新 Markdown 内容",
|
||||||
"split_on_marker": "连续短消息",
|
"split_on_marker": "连续短消息",
|
||||||
"split_on_marker_hint": "像真人聊天一样,把长难句拆成多条短消息快速发出",
|
"split_on_marker_hint": "像真人聊天一样,把长难句拆成多条短消息快速发出",
|
||||||
"tool_feedback_enabled": "工具反馈",
|
"tool_feedback_enabled": "工具反馈",
|
||||||
|
|
|
||||||
|
|
@ -37,6 +37,7 @@ export interface ChatMessage {
|
||||||
content: string
|
content: string
|
||||||
timestamp: number | string
|
timestamp: number | string
|
||||||
kind?: AssistantMessageKind
|
kind?: AssistantMessageKind
|
||||||
|
streaming?: boolean
|
||||||
attachments?: ChatAttachment[]
|
attachments?: ChatAttachment[]
|
||||||
toolCalls?: ChatToolCall[]
|
toolCalls?: ChatToolCall[]
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue