refactor(streaming): remove dead code and simplify streaming wiring
- Delete unused Anthropic ChatStream/parseStream (-131 lines) — factory creates HTTPProvider for all OpenAI-compat providers including OpenRouter - Simplify runLLMIteration from 4 to 3 return values (remove unused streamed bool) - Replace managerStreamer struct with finalizeHookStreamer using embedding (Update/Cancel promoted, only Finalize overridden)
This commit is contained in:
Amir Mamaghani
parent
6ab2c8239d
commit
9792df6fda
3 changed files with 23 additions and 171 deletions
|
|
@ -623,7 +623,7 @@ func (al *AgentLoop) runAgentLoop(
|
|||
agent.Sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage)
|
||||
|
||||
// 4. Run LLM iteration loop
|
||||
finalContent, iteration, streamSent, err := al.runLLMIteration(ctx, agent, messages, opts)
|
||||
finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
|
@ -645,8 +645,11 @@ func (al *AgentLoop) runAgentLoop(
|
|||
al.maybeSummarize(agent, opts.SessionKey, opts.Channel, opts.ChatID)
|
||||
}
|
||||
|
||||
// 8. Optional: send response via bus (skip if streaming already delivered it)
|
||||
if opts.SendResponse && !streamSent {
|
||||
// 8. Optional: send response via bus.
|
||||
// Even when streaming already delivered the message, we still publish so that
|
||||
// preSend can clean up typing/reaction/placeholder. The streamActive flag in
|
||||
// the manager's preSend will skip the actual Send (no duplicate).
|
||||
if opts.SendResponse {
|
||||
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
|
||||
Channel: opts.Channel,
|
||||
ChatID: opts.ChatID,
|
||||
|
|
@ -664,11 +667,6 @@ func (al *AgentLoop) runAgentLoop(
|
|||
"final_length": len(finalContent),
|
||||
})
|
||||
|
||||
// When streaming already delivered the message, return empty so the caller
|
||||
// (Run loop) doesn't publish a duplicate via PublishOutbound.
|
||||
if streamSent {
|
||||
return "", nil
|
||||
}
|
||||
return finalContent, nil
|
||||
}
|
||||
|
||||
|
|
@ -729,18 +727,15 @@ func (al *AgentLoop) handleReasoning(
|
|||
}
|
||||
|
||||
// runLLMIteration executes the LLM call loop with tool handling.
|
||||
// Returns (finalContent, iteration, streamed, error).
|
||||
// When streamed is true, the response was already delivered to the user via
|
||||
// streaming (sendMessageDraft + sendMessage) and PublishOutbound should be skipped.
|
||||
// Returns (finalContent, iteration, error).
|
||||
func (al *AgentLoop) runLLMIteration(
|
||||
ctx context.Context,
|
||||
agent *AgentInstance,
|
||||
messages []providers.Message,
|
||||
opts processOptions,
|
||||
) (string, int, bool, error) {
|
||||
) (string, int, error) {
|
||||
iteration := 0
|
||||
var finalContent string
|
||||
var streamed bool
|
||||
|
||||
// Check if both the provider and channel support streaming
|
||||
streamProvider, providerCanStream := agent.Provider.(providers.StreamingProvider)
|
||||
|
|
@ -906,7 +901,7 @@ func (al *AgentLoop) runLLMIteration(
|
|||
"iteration": iteration,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", iteration, false, fmt.Errorf("LLM call failed after retries: %w", err)
|
||||
return "", iteration, fmt.Errorf("LLM call failed after retries: %w", err)
|
||||
}
|
||||
|
||||
go al.handleReasoning(
|
||||
|
|
@ -936,8 +931,6 @@ func (al *AgentLoop) runLLMIteration(
|
|||
logger.WarnCF("agent", "Stream finalize failed", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
} else {
|
||||
streamed = true
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -946,7 +939,7 @@ func (al *AgentLoop) runLLMIteration(
|
|||
"agent_id": agent.ID,
|
||||
"iteration": iteration,
|
||||
"content_chars": len(finalContent),
|
||||
"streamed": streamed,
|
||||
"streamed": streamer != nil,
|
||||
})
|
||||
break
|
||||
}
|
||||
|
|
@ -1111,7 +1104,7 @@ func (al *AgentLoop) runLLMIteration(
|
|||
}
|
||||
}
|
||||
|
||||
return finalContent, iteration, streamed, nil
|
||||
return finalContent, iteration, nil
|
||||
}
|
||||
|
||||
// updateToolContexts updates the context for tools that need channel/chatID info.
|
||||
|
|
|
|||
|
|
@ -207,32 +207,23 @@ func (m *Manager) GetStreamer(ctx context.Context, channelName, chatID string) (
|
|||
return nil, false
|
||||
}
|
||||
|
||||
// Wrap the channels.Streamer to track finalization for preSend coordination
|
||||
return &managerStreamer{
|
||||
inner: streamer,
|
||||
manager: m,
|
||||
key: channelName + ":" + chatID,
|
||||
// Mark streamActive on Finalize so preSend knows to clean up the placeholder
|
||||
key := channelName + ":" + chatID
|
||||
return &finalizeHookStreamer{
|
||||
Streamer: streamer,
|
||||
onFinalize: func() { m.streamActive.Store(key, true) },
|
||||
}, true
|
||||
}
|
||||
|
||||
// managerStreamer wraps a channels.Streamer to mark streamActive on Finalize.
|
||||
type managerStreamer struct {
|
||||
inner Streamer
|
||||
manager *Manager
|
||||
key string
|
||||
// finalizeHookStreamer wraps a Streamer to run a hook on Finalize.
|
||||
type finalizeHookStreamer struct {
|
||||
Streamer
|
||||
onFinalize func()
|
||||
}
|
||||
|
||||
func (s *managerStreamer) Update(ctx context.Context, content string) error {
|
||||
return s.inner.Update(ctx, content)
|
||||
}
|
||||
|
||||
func (s *managerStreamer) Finalize(ctx context.Context, content string) error {
|
||||
s.manager.streamActive.Store(s.key, true)
|
||||
return s.inner.Finalize(ctx, content)
|
||||
}
|
||||
|
||||
func (s *managerStreamer) Cancel(ctx context.Context) {
|
||||
s.inner.Cancel(ctx)
|
||||
func (s *finalizeHookStreamer) Finalize(ctx context.Context, content string) error {
|
||||
s.onFinalize()
|
||||
return s.Streamer.Finalize(ctx, content)
|
||||
}
|
||||
|
||||
// initChannel is a helper that looks up a factory by name and creates the channel.
|
||||
|
|
|
|||
|
|
@ -9,8 +9,6 @@ import (
|
|||
|
||||
"github.com/anthropics/anthropic-sdk-go"
|
||||
"github.com/anthropics/anthropic-sdk-go/option"
|
||||
"github.com/anthropics/anthropic-sdk-go/packages/ssestream"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/providers/protocoltypes"
|
||||
)
|
||||
|
||||
|
|
@ -94,136 +92,6 @@ func (p *Provider) Chat(
|
|||
return parseResponse(resp), nil
|
||||
}
|
||||
|
||||
// ChatStream implements providers.StreamingProvider.
|
||||
// It streams text deltas to onChunk (with accumulated text) while building
|
||||
// the same LLMResponse returned by Chat for tool-call compatibility.
|
||||
func (p *Provider) ChatStream(
|
||||
ctx context.Context,
|
||||
messages []Message,
|
||||
tools []ToolDefinition,
|
||||
model string,
|
||||
options map[string]any,
|
||||
onChunk func(accumulated string),
|
||||
) (*LLMResponse, error) {
|
||||
var opts []option.RequestOption
|
||||
if p.tokenSource != nil {
|
||||
tok, err := p.tokenSource()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("refreshing token: %w", err)
|
||||
}
|
||||
opts = append(opts, option.WithAuthToken(tok))
|
||||
}
|
||||
|
||||
params, err := buildParams(messages, tools, model, options)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
stream := p.client.Messages.NewStreaming(ctx, params, opts...)
|
||||
return parseStream(stream, onChunk)
|
||||
}
|
||||
|
||||
// parseStream consumes a streaming response and builds an LLMResponse.
|
||||
func parseStream(
|
||||
stream *ssestream.Stream[anthropic.MessageStreamEventUnion],
|
||||
onChunk func(accumulated string),
|
||||
) (*LLMResponse, error) {
|
||||
defer stream.Close()
|
||||
|
||||
var textContent strings.Builder
|
||||
var toolCalls []ToolCall
|
||||
var stopReason anthropic.StopReason
|
||||
var inputTokens, outputTokens int64
|
||||
|
||||
// Track tool_use blocks being assembled (by content block index)
|
||||
type toolBlock struct {
|
||||
id string
|
||||
name string
|
||||
inputJSON strings.Builder
|
||||
}
|
||||
activeTools := map[int64]*toolBlock{}
|
||||
|
||||
for stream.Next() {
|
||||
event := stream.Current()
|
||||
|
||||
switch event.Type {
|
||||
case "message_start":
|
||||
if event.Message.Usage.InputTokens > 0 {
|
||||
inputTokens = event.Message.Usage.InputTokens
|
||||
}
|
||||
|
||||
case "content_block_start":
|
||||
cb := event.ContentBlock
|
||||
if cb.Type == "tool_use" {
|
||||
activeTools[event.Index] = &toolBlock{
|
||||
id: cb.ID,
|
||||
name: cb.Name,
|
||||
}
|
||||
}
|
||||
|
||||
case "content_block_delta":
|
||||
delta := event.Delta
|
||||
switch delta.Type {
|
||||
case "text_delta":
|
||||
textContent.WriteString(delta.Text)
|
||||
if onChunk != nil {
|
||||
onChunk(textContent.String())
|
||||
}
|
||||
case "input_json_delta":
|
||||
if tb, ok := activeTools[event.Index]; ok {
|
||||
tb.inputJSON.WriteString(delta.PartialJSON)
|
||||
}
|
||||
}
|
||||
|
||||
case "content_block_stop":
|
||||
if tb, ok := activeTools[event.Index]; ok {
|
||||
var args map[string]any
|
||||
if err := json.Unmarshal([]byte(tb.inputJSON.String()), &args); err != nil {
|
||||
log.Printf("anthropic stream: failed to decode tool call input for %q: %v", tb.name, err)
|
||||
args = map[string]any{"raw": tb.inputJSON.String()}
|
||||
}
|
||||
toolCalls = append(toolCalls, ToolCall{
|
||||
ID: tb.id,
|
||||
Name: tb.name,
|
||||
Arguments: args,
|
||||
})
|
||||
delete(activeTools, event.Index)
|
||||
}
|
||||
|
||||
case "message_delta":
|
||||
stopReason = event.Delta.StopReason
|
||||
if event.Usage.OutputTokens > 0 {
|
||||
outputTokens = event.Usage.OutputTokens
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err := stream.Err(); err != nil {
|
||||
return nil, fmt.Errorf("claude streaming API call: %w", err)
|
||||
}
|
||||
|
||||
finishReason := "stop"
|
||||
switch stopReason {
|
||||
case anthropic.StopReasonToolUse:
|
||||
finishReason = "tool_calls"
|
||||
case anthropic.StopReasonMaxTokens:
|
||||
finishReason = "length"
|
||||
case anthropic.StopReasonEndTurn:
|
||||
finishReason = "stop"
|
||||
}
|
||||
|
||||
return &LLMResponse{
|
||||
Content: textContent.String(),
|
||||
ToolCalls: toolCalls,
|
||||
FinishReason: finishReason,
|
||||
Usage: &UsageInfo{
|
||||
PromptTokens: int(inputTokens),
|
||||
CompletionTokens: int(outputTokens),
|
||||
TotalTokens: int(inputTokens + outputTokens),
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (p *Provider) GetDefaultModel() string {
|
||||
return "claude-sonnet-4.6"
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue