weichen
3b92cb63b4
Why: Interleaved-thinking models (MiniMax, DeepSeek R1) require the assistant's thinking to be passed back verbatim in message history; dropping it degrades later turns. The product's P2-I1 replayed thinking only into the request-side rebuild; with the session transcript becoming the authoritative context (quality P2-I2), capture must live in the kernel so thinking persists in the transcript and replays on every request. Evidence: SELF - TestReasoningCaptureAndReplay covers capture (SSE reasoning_content into Message.Thinking, content unpolluted) and replay (<think> prefix on thinking assistants, untouched non-thinking messages); go vet + go test ./... pass. Impact: sse.go accumulates reasoning deltas into the terminal message's Thinking; client.go toProviderMessages renders captured thinking as a <think> block prefix for assistant messages. Anthropic path unchanged; messages without thinking serialize exactly as before.
405 lines
13 KiB
Go
405 lines
13 KiB
Go
package llm
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
// anthropic.go — 原生 Anthropic Messages API 客户端（N2 WP2）。
|
||
// 与 OpenAI 路径共用 Message/Event 表面；组包与 BuildBody 的 Anthropic
|
||
// 分支是同一个函数（消灭「离线/线上两张皮」，审计 N4 的根因）。
|
||
|
||
// AnthropicVersion 是请求头 anthropic-version 的钉住值。
|
||
const AnthropicVersion = "2023-06-01"
|
||
|
||
// anthropicBody 组 Anthropic Messages 请求体（线上 Client 与离线 BuildBody
|
||
// 共用）。tools 挂 input_schema；system 与最后一个 tool 挂 cache_control
|
||
// （静态前缀的缓存断点）；assistant 的思考块在内容块最前且带签名往返。
|
||
func anthropicBody(model string, messages []Message, tools []ToolDecl, opts StreamOptions) ([]byte, error) {
|
||
sys, msgs := anthropicMessages(messages)
|
||
body := map[string]any{
|
||
"model": model,
|
||
"messages": msgs,
|
||
"stream": true,
|
||
// Anthropic 要求 max_tokens。0 表示调用方不限制：用足够大的默认值，
|
||
// 避免 4096 把长回复截断。真正的上限由调用方 StreamOptions.MaxTokens 决定。
|
||
"max_tokens": opts.MaxTokens,
|
||
}
|
||
if body["max_tokens"] == 0 {
|
||
body["max_tokens"] = 128000
|
||
}
|
||
if sys != "" {
|
||
body["system"] = []map[string]any{{
|
||
"type": "text",
|
||
"text": sys,
|
||
"cache_control": map[string]any{"type": "ephemeral"},
|
||
}}
|
||
}
|
||
if budget := AnthropicThinkingBudget(opts.ThinkingLevel); budget > 0 {
|
||
body["thinking"] = map[string]any{"type": "enabled", "budget_tokens": budget}
|
||
}
|
||
if len(tools) > 0 {
|
||
body["tools"] = anthropicTools(tools)
|
||
}
|
||
return json.Marshal(body)
|
||
}
|
||
|
||
// anthropicTools 声明工具（name/description/input_schema）；最后一个挂
|
||
// cache_control 作为静态工具前缀的断点。
|
||
func anthropicTools(decls []ToolDecl) []map[string]any {
|
||
out := make([]map[string]any, 0, len(decls))
|
||
for i, t := range decls {
|
||
params := t.Parameters
|
||
if len(params) == 0 {
|
||
params = json.RawMessage(`{"type":"object","properties":{}}`)
|
||
}
|
||
item := map[string]any{
|
||
"name": t.Name,
|
||
"description": t.Description,
|
||
"input_schema": json.RawMessage(params),
|
||
}
|
||
if i == len(decls)-1 {
|
||
item["cache_control"] = map[string]any{"type": "ephemeral"}
|
||
}
|
||
out = append(out, item)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// anthropicMessages 把通用转录映射成 Anthropic 消息：
|
||
// system 收成字符串（由调用方决定挂法）；assistant 的思考块在最前，
|
||
// 其后 text/image 与 tool_use；连续 RoleTool 折叠成一条 user 消息内的
|
||
// tool_result 块组（Anthropic 不允许 tool 角色）。
|
||
func anthropicMessages(messages []Message) (string, []map[string]any) {
|
||
var system strings.Builder
|
||
var out []map[string]any
|
||
for i := 0; i < len(messages); i++ {
|
||
m := messages[i]
|
||
switch m.Role {
|
||
case RoleSystem:
|
||
if system.Len() > 0 {
|
||
system.WriteString("\n")
|
||
}
|
||
system.WriteString(m.Content)
|
||
case RoleAssistant:
|
||
var content []map[string]any
|
||
for _, th := range m.Thinking {
|
||
if th.Redacted {
|
||
content = append(content, map[string]any{"type": "redacted_thinking", "data": th.Data})
|
||
continue
|
||
}
|
||
content = append(content, map[string]any{
|
||
"type": "thinking", "thinking": th.Thinking, "signature": th.Signature,
|
||
})
|
||
}
|
||
if m.Content != "" {
|
||
content = append(content, map[string]any{"type": "text", "text": m.Content})
|
||
}
|
||
for _, img := range m.Images {
|
||
content = append(content, map[string]any{
|
||
"type": "image",
|
||
"source": map[string]any{
|
||
"type": "base64",
|
||
"media_type": firstNonEmpty(img.MIME, "image/png"),
|
||
"data": img.Data,
|
||
},
|
||
})
|
||
}
|
||
for _, tc := range m.ToolCalls {
|
||
args := json.RawMessage(tc.Arguments)
|
||
if !json.Valid(args) {
|
||
args = json.RawMessage(`{}`)
|
||
}
|
||
content = append(content, map[string]any{
|
||
"type": "tool_use", "id": tc.ID, "name": tc.Name, "input": args,
|
||
})
|
||
}
|
||
if len(content) == 0 {
|
||
content = append(content, map[string]any{"type": "text", "text": ""})
|
||
}
|
||
out = append(out, map[string]any{"role": "assistant", "content": content})
|
||
case RoleTool:
|
||
// 连续 tool 结果折叠进同一条 user 消息。
|
||
var results []map[string]any
|
||
for ; i < len(messages) && messages[i].Role == RoleTool; i++ {
|
||
t := messages[i]
|
||
results = append(results, map[string]any{
|
||
"type": "tool_result",
|
||
"tool_use_id": t.ToolCallID,
|
||
"content": t.Content,
|
||
"is_error": t.IsError,
|
||
})
|
||
}
|
||
i--
|
||
out = append(out, map[string]any{"role": "user", "content": results})
|
||
default:
|
||
content := []map[string]any{}
|
||
if m.Content != "" {
|
||
content = append(content, map[string]any{"type": "text", "text": m.Content})
|
||
}
|
||
for _, img := range m.Images {
|
||
content = append(content, map[string]any{
|
||
"type": "image",
|
||
"source": map[string]any{
|
||
"type": "base64",
|
||
"media_type": firstNonEmpty(img.MIME, "image/png"),
|
||
"data": img.Data,
|
||
},
|
||
})
|
||
}
|
||
out = append(out, map[string]any{"role": "user", "content": content})
|
||
}
|
||
}
|
||
return system.String(), out
|
||
}
|
||
|
||
// streamAnthropic 发 POST {base}/messages 并把 Anthropic SSE 事件族映射到
|
||
// 通用 Event。与 OpenAI 客户端共享错误/中止语义。
|
||
func (c *Client) streamAnthropic(ctx context.Context, model string, messages []Message, tools []ToolDecl, opts StreamOptions, out chan<- Event) {
|
||
base := firstNonEmpty(opts.APIBase, c.BaseURL)
|
||
key := firstNonEmpty(opts.APIKey, c.APIKey)
|
||
if base == "" || key == "" || model == "" {
|
||
emitError(out, "", false, "agentx llm client is missing base url, api key, or model")
|
||
return
|
||
}
|
||
if opts.MessageShape != "" {
|
||
messages = PrepareMessages(messages, opts.MessageShape)
|
||
}
|
||
raw, err := anthropicBody(model, messages, tools, opts)
|
||
if err != nil {
|
||
emitError(out, "", false, err.Error())
|
||
return
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(base, "/")+"/messages", bytes.NewReader(raw))
|
||
if err != nil {
|
||
emitError(out, "", false, err.Error())
|
||
return
|
||
}
|
||
req.Header.Set("x-api-key", key)
|
||
req.Header.Set("anthropic-version", AnthropicVersion)
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("Accept", "text/event-stream")
|
||
hc := c.HTTPClient
|
||
if hc == nil {
|
||
hc = &http.Client{Timeout: 30 * time.Minute}
|
||
}
|
||
resp, err := hc.Do(req)
|
||
if err != nil {
|
||
if ctx.Err() != nil {
|
||
emitAbort(out, "", ctx.Err())
|
||
return
|
||
}
|
||
emitError(out, "", true, err.Error())
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode >= 400 {
|
||
slurp, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
||
msg := fmt.Sprintf("provider status %d: %s", resp.StatusCode, strings.TrimSpace(string(slurp)))
|
||
ev := Event{
|
||
Kind: KindError,
|
||
HTTPStatus: resp.StatusCode,
|
||
Retryable: StatusRetryable(resp.StatusCode) && !ContextOverflow(msg),
|
||
ErrorMessage: msg,
|
||
Err: &StreamError{Message: msg, Retryable: StatusRetryable(resp.StatusCode) && !ContextOverflow(msg)},
|
||
Message: Message{Role: RoleAssistant, StopReason: StopError},
|
||
}
|
||
if d, ok := ParseRetryAfter(resp.Header.Get("Retry-After"), time.Now()); ok {
|
||
ev.HasRetryAfter = true
|
||
ev.RetryAfter = d
|
||
}
|
||
out <- ev
|
||
return
|
||
}
|
||
parseAnthropicSSE(ctx, resp.Body, out)
|
||
}
|
||
|
||
// ---- Anthropic SSE 解析 ----
|
||
|
||
type anthropicBlock struct {
|
||
typ string
|
||
toolID string
|
||
toolName string
|
||
args strings.Builder
|
||
thinking strings.Builder
|
||
signature strings.Builder
|
||
redacted string
|
||
}
|
||
|
||
// parseAnthropicSSE 读 event:/data: 帧对，产出文本/思考/工具增量与一次
|
||
// 终结 message_end（或 error）。
|
||
func parseAnthropicSSE(ctx context.Context, r io.Reader, out chan<- Event) {
|
||
blocks := map[int]*anthropicBlock{}
|
||
var content strings.Builder
|
||
var thinking []ThinkingBlock
|
||
var calls []ToolCall
|
||
var usage Usage
|
||
stopReason := StopEnd
|
||
finishReason := ""
|
||
sc := bufio.NewScanner(r)
|
||
sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
|
||
eventType := ""
|
||
for sc.Scan() {
|
||
line := sc.Text()
|
||
select {
|
||
case <-ctx.Done():
|
||
emitAbort(out, "", ctx.Err())
|
||
return
|
||
default:
|
||
}
|
||
switch {
|
||
case strings.HasPrefix(line, "event:"):
|
||
eventType = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
|
||
case strings.HasPrefix(line, "data:"):
|
||
data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
|
||
if data == "" {
|
||
continue
|
||
}
|
||
var frame map[string]json.RawMessage
|
||
if err := json.Unmarshal([]byte(data), &frame); err != nil {
|
||
continue
|
||
}
|
||
switch eventType {
|
||
case "message_start":
|
||
var v struct {
|
||
Message struct {
|
||
Usage struct {
|
||
InputTokens int `json:"input_tokens"`
|
||
CacheReadTokens int `json:"cache_read_input_tokens"`
|
||
CacheWriteTokens int `json:"cache_creation_input_tokens"`
|
||
} `json:"usage"`
|
||
} `json:"message"`
|
||
}
|
||
if json.Unmarshal(frame["message"], &v.Message) == nil {
|
||
usage.PromptTokens = v.Message.Usage.InputTokens
|
||
usage.CacheReadTokens = v.Message.Usage.CacheReadTokens
|
||
usage.CacheWriteTokens = v.Message.Usage.CacheWriteTokens
|
||
}
|
||
case "content_block_start":
|
||
var v struct {
|
||
Index int `json:"index"`
|
||
ContentBlock struct {
|
||
Type string `json:"type"`
|
||
ID string `json:"id"`
|
||
Name string `json:"name"`
|
||
Data string `json:"data"`
|
||
} `json:"content_block"`
|
||
}
|
||
if json.Unmarshal(frame["index"], &v.Index) == nil {
|
||
_ = json.Unmarshal(frame["content_block"], &v.ContentBlock)
|
||
b := &anthropicBlock{typ: v.ContentBlock.Type, toolID: v.ContentBlock.ID, toolName: v.ContentBlock.Name, redacted: v.ContentBlock.Data}
|
||
blocks[v.Index] = b
|
||
if v.ContentBlock.Type == "tool_use" {
|
||
out <- Event{Kind: KindToolCallDelta, ToolIndex: v.Index, ToolID: v.ContentBlock.ID, ToolName: v.ContentBlock.Name}
|
||
}
|
||
}
|
||
case "content_block_delta":
|
||
var v struct {
|
||
Index int `json:"index"`
|
||
Delta struct {
|
||
Type string `json:"type"`
|
||
Text string `json:"text"`
|
||
Thinking string `json:"thinking"`
|
||
Signature string `json:"signature"`
|
||
Partial string `json:"partial_json"`
|
||
} `json:"delta"`
|
||
}
|
||
if json.Unmarshal(frame["index"], &v.Index) != nil {
|
||
continue
|
||
}
|
||
_ = json.Unmarshal(frame["delta"], &v.Delta)
|
||
b := blocks[v.Index]
|
||
switch v.Delta.Type {
|
||
case "text_delta":
|
||
content.WriteString(v.Delta.Text)
|
||
out <- Event{Kind: KindTextDelta, Text: v.Delta.Text}
|
||
case "thinking_delta":
|
||
if b != nil {
|
||
b.thinking.WriteString(v.Delta.Thinking)
|
||
}
|
||
out <- Event{Kind: KindReasoningDelta, ReasoningDelta: v.Delta.Thinking}
|
||
case "signature_delta":
|
||
if b != nil {
|
||
b.signature.WriteString(v.Delta.Signature)
|
||
}
|
||
case "input_json_delta":
|
||
if b != nil {
|
||
b.args.WriteString(v.Delta.Partial)
|
||
}
|
||
out <- Event{Kind: KindToolCallDelta, ToolIndex: v.Index, ArgumentsDelta: v.Delta.Partial}
|
||
}
|
||
case "content_block_stop":
|
||
var v struct {
|
||
Index int `json:"index"`
|
||
}
|
||
if json.Unmarshal(frame["index"], &v.Index) != nil {
|
||
continue
|
||
}
|
||
b := blocks[v.Index]
|
||
if b == nil {
|
||
continue
|
||
}
|
||
switch b.typ {
|
||
case "tool_use":
|
||
calls = append(calls, ToolCall{ID: b.toolID, Name: b.toolName, Arguments: b.args.String()})
|
||
case "thinking":
|
||
thinking = append(thinking, ThinkingBlock{Thinking: b.thinking.String(), Signature: b.signature.String()})
|
||
case "redacted_thinking":
|
||
thinking = append(thinking, ThinkingBlock{Redacted: true, Data: b.redacted})
|
||
}
|
||
case "message_delta":
|
||
var v struct {
|
||
Delta struct {
|
||
StopReason string `json:"stop_reason"`
|
||
} `json:"delta"`
|
||
Usage struct {
|
||
OutputTokens int `json:"output_tokens"`
|
||
} `json:"usage"`
|
||
}
|
||
_ = json.Unmarshal(frame["delta"], &v.Delta)
|
||
_ = json.Unmarshal(frame["usage"], &v.Usage)
|
||
finishReason = v.Delta.StopReason
|
||
usage.CompletionTokens = v.Usage.OutputTokens
|
||
if v.Delta.StopReason == "tool_use" {
|
||
stopReason = StopToolCalls
|
||
}
|
||
case "error":
|
||
msg := data
|
||
var v struct {
|
||
Error struct {
|
||
Type string `json:"type"`
|
||
Message string `json:"message"`
|
||
} `json:"error"`
|
||
}
|
||
if json.Unmarshal(frame["error"], &v.Error) == nil && v.Error.Message != "" {
|
||
msg = v.Error.Type + ": " + v.Error.Message
|
||
}
|
||
out <- Event{Kind: KindError, ErrorMessage: msg, Err: &StreamError{Message: msg}, Message: Message{Role: RoleAssistant, StopReason: StopError}}
|
||
return
|
||
}
|
||
}
|
||
}
|
||
if err := sc.Err(); err != nil {
|
||
if ctx.Err() != nil {
|
||
emitAbort(out, "", ctx.Err())
|
||
return
|
||
}
|
||
emitError(out, "", true, "anthropic sse read: "+err.Error())
|
||
return
|
||
}
|
||
usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens
|
||
out <- Event{
|
||
Kind: KindMessageEnd,
|
||
Message: Message{Role: RoleAssistant, Content: content.String(), ToolCalls: calls, Thinking: thinking, StopReason: stopReason},
|
||
Usage: usage,
|
||
FinishReason: finishReason,
|
||
}
|
||
}
|