diff --git a/runtime/llm/anthropic.go b/runtime/llm/anthropic.go new file mode 100644 index 0000000..931156a --- /dev/null +++ b/runtime/llm/anthropic.go @@ -0,0 +1,404 @@ +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, + } +} diff --git a/runtime/llm/anthropic_test.go b/runtime/llm/anthropic_test.go new file mode 100644 index 0000000..267f4c0 --- /dev/null +++ b/runtime/llm/anthropic_test.go @@ -0,0 +1,250 @@ +package llm + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" +) + +// ---- N2 WP2:Anthropic 原生客户端(验收① 的机器证据)---- + +// fixtureServer 按调用序回放 testdata/anthropic 下的录制 SSE,并记录请求。 +type fixtureServer struct { + t *testing.T + requests []recordedRequest + count int +} + +type recordedRequest struct { + path string + apiKey string + version string + body map[string]any +} + +func (f *fixtureServer) handler(w http.ResponseWriter, r *http.Request) { + f.count++ + body, _ := io.ReadAll(r.Body) + var parsed map[string]any + _ = json.Unmarshal(body, &parsed) + f.requests = append(f.requests, recordedRequest{ + path: r.URL.Path, + apiKey: r.Header.Get("x-api-key"), + version: r.Header.Get("anthropic-version"), + body: parsed, + }) + names := []string{"01-tool-use.sse", "02-thinking-text.sse", "03-text.sse"} + raw, err := os.ReadFile(filepath.Join("testdata", "anthropic", names[f.count-1])) + if err != nil { + f.t.Fatalf("fixture %d: %v", f.count, err) + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write(raw) +} + +// anthropicPresetClient 经预设表构建客户端(验收的是「选 anthropic 预设」本身), +// BaseURL 指到夹具服务器(预设只是结构体,测试内覆盖端点,不打真实网络)。 +func anthropicPresetClient(t *testing.T, base string) *Client { + t.Helper() + var preset Provider + for _, p := range Presets() { + if p.Name == "anthropic" { + preset = p + } + } + if preset.Name == "" { + t.Fatal("anthropic preset missing") + } + preset.BaseURL = base + preset.APIKey = "fixture-key" + c, err := ClientFor(preset) + if err != nil { + t.Fatalf("ClientFor: %v", err) + } + return c +} + +func collect(t *testing.T, ch <-chan Event) (end Event, texts, reasoning []string) { + t.Helper() + for ev := range ch { + switch ev.Kind { + case KindTextDelta: + texts = append(texts, ev.Text) + case KindReasoningDelta: + reasoning = append(reasoning, ev.ReasoningDelta) + case KindMessageEnd: + end = ev + case KindError: + t.Fatalf("stream error: %s", ev.ErrorMessage) + } + } + return end, texts, reasoning +} + +// TestAnthropicPresetToolRound 验收①:选 anthropic 预设,用录制 HTTP 夹具 +// 完成一轮完整工具调用(tools 请求 → tool_use → tool_result → 终文本), +// 且 thinking 块带原签名在下一请求里往返。 +func TestAnthropicPresetToolRound(t *testing.T) { + fs := &fixtureServer{t: t} + srv := httptest.NewServer(http.HandlerFunc(fs.handler)) + defer srv.Close() + c := anthropicPresetClient(t, srv.URL) + ctx := context.Background() + + tools := []ToolDecl{{Name: "read_file", Description: "读文件", Parameters: json.RawMessage(`{"type":"object","properties":{"path":{"type":"string"}}}`)}} + + // 第一回合:带 tools 请求 → tool_use。 + end1, _, _ := collect(t, c.Stream(ctx, "claude-fixture", []Message{{Role: RoleUser, Content: "查一下 main.go"}}, tools, StreamOptions{})) + if end1.Message.StopReason != StopToolCalls || len(end1.Message.ToolCalls) != 1 { + t.Fatalf("resp1: %+v", end1.Message) + } + call := end1.Message.ToolCalls[0] + if call.ID != "toolu_01ABC" || call.Name != "read_file" || call.Arguments != `{"path":"main.go"}` { + t.Fatalf("tool call: %+v", call) + } + + req1 := fs.requests[0] + if req1.path != "/messages" || req1.apiKey != "fixture-key" || req1.version != AnthropicVersion { + t.Fatalf("req1 headers: %+v", req1) + } + toolsBody, _ := req1.body["tools"].([]any) + if len(toolsBody) != 1 { + t.Fatalf("req1 tools: %v", req1.body["tools"]) + } + tool0 := toolsBody[0].(map[string]any) + if tool0["name"] != "read_file" || tool0["input_schema"] == nil || tool0["cache_control"] == nil { + t.Fatalf("req1 tool decl: %v", tool0) + } + + // 第二回合:assistant(tool_use) + tool_result → thinking(带签名) + 终文本。 + msgs2 := []Message{ + {Role: RoleUser, Content: "查一下 main.go"}, + end1.Message, + {Role: RoleTool, ToolCallID: call.ID, Name: call.Name, Content: "1 package main\n..."}, + } + end2, texts2, reasoning2 := collect(t, c.Stream(ctx, "claude-fixture", msgs2, tools, StreamOptions{})) + if end2.Message.StopReason != StopEnd || end2.Message.Content != "查到了,main.go 一共 42 行。" { + t.Fatalf("resp2: %+v", end2.Message) + } + if len(end2.Message.Thinking) != 1 || end2.Message.Thinking[0].Signature != "EqoBCkgIARABGAIiQA" { + t.Fatalf("resp2 thinking: %+v", end2.Message.Thinking) + } + if strings.Join(texts2, "") != end2.Message.Content || len(reasoning2) == 0 { + t.Fatalf("deltas: %v %v", texts2, reasoning2) + } + + req2 := fs.requests[1] + msgsBody, _ := req2.body["messages"].([]any) + var toolResult map[string]any + for _, m := range msgsBody { + mm := m.(map[string]any) + if mm["role"] != "user" { + continue + } + for _, blk := range mm["content"].([]any) { + b := blk.(map[string]any) + if b["type"] == "tool_result" { + toolResult = b + } + } + } + if toolResult == nil || toolResult["tool_use_id"] != "toolu_01ABC" { + t.Fatalf("req2 tool_result: %v", toolResult) + } + + // 第三回合:assistant(thinking+text) 回送 → 签名必须原样往返。 + msgs3 := append(append([]Message{}, msgs2...), end2.Message, Message{Role: RoleUser, Content: "再说一遍"}) + end3, _, _ := collect(t, c.Stream(ctx, "claude-fixture", msgs3, tools, StreamOptions{})) + if end3.Message.Content != "42 行。" { + t.Fatalf("resp3: %+v", end3.Message) + } + + req3 := fs.requests[2] + var thinkingBlock map[string]any + for _, m := range req3.body["messages"].([]any) { + mm := m.(map[string]any) + if mm["role"] != "assistant" { + continue + } + for _, blk := range mm["content"].([]any) { + b := blk.(map[string]any) + if b["type"] == "thinking" { + thinkingBlock = b + } + } + } + if thinkingBlock == nil || thinkingBlock["signature"] != "EqoBCkgIARABGAIiQA" { + t.Fatalf("req3 thinking round-trip: %v", thinkingBlock) + } +} + +// TestAnthropicBodyShape 组包单测:tool_use/tool_result/thinking 块形态、 +// system 与末位工具的 cache_control、max_tokens 默认。 +func TestAnthropicBodyShape(t *testing.T) { + msgs := []Message{ + {Role: RoleSystem, Content: "sys"}, + {Role: RoleUser, Content: "u"}, + {Role: RoleAssistant, Thinking: []ThinkingBlock{{Thinking: "想", Signature: "sig"}}, ToolCalls: []ToolCall{{ID: "t1", Name: "read", Arguments: `{"a":1}`}}}, + {Role: RoleTool, ToolCallID: "t1", Name: "read", Content: "r1"}, + {Role: RoleTool, ToolCallID: "t2", Name: "grep", Content: "r2", IsError: true}, + } + raw, err := anthropicBody("claude", msgs, []ToolDecl{{Name: "read", Description: "读"}}, StreamOptions{}) + if err != nil { + t.Fatal(err) + } + var body map[string]any + if err := json.Unmarshal(raw, &body); err != nil { + t.Fatal(err) + } + if body["max_tokens"] != float64(128000) { + t.Fatalf("max_tokens 默认:%v", body["max_tokens"]) + } + sys := body["system"].([]any)[0].(map[string]any) + if sys["cache_control"] == nil { + t.Fatal("system 应挂 cache_control") + } + out := body["messages"].([]any) + // assistant:思考块在最前,其后 tool_use。 + assistant := out[1].(map[string]any) + blocks := assistant["content"].([]any) + if blocks[0].(map[string]any)["type"] != "thinking" || blocks[1].(map[string]any)["type"] != "tool_use" { + t.Fatalf("assistant 块序:%v", blocks) + } + // 连续两条 tool 结果折叠进同一条 user 消息。 + folded := out[2].(map[string]any) + if folded["role"] != "user" || len(folded["content"].([]any)) != 2 { + t.Fatalf("tool_result 折叠:%v", folded) + } + second := folded["content"].([]any)[1].(map[string]any) + if second["type"] != "tool_result" || second["is_error"] != true { + t.Fatalf("tool_result is_error:%v", second) + } + // 末位工具挂 cache_control。 + tool := body["tools"].([]any)[0].(map[string]any) + if tool["cache_control"] == nil { + t.Fatal("末位工具应挂 cache_control") + } +} + +// TestAnthropicSSEErrorFrame 协议内 error 帧映射为 KindError。 +func TestAnthropicSSEErrorFrame(t *testing.T) { + raw := "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}\n\n" + ch := make(chan Event, 1) + go func() { + defer close(ch) + parseAnthropicSSE(context.Background(), strings.NewReader(raw), ch) + }() + var got []Event + for ev := range ch { + got = append(got, ev) + } + if len(got) != 1 || got[0].Kind != KindError || !strings.Contains(got[0].ErrorMessage, "Overloaded") { + t.Fatalf("%+v", got) + } +} diff --git a/runtime/llm/client.go b/runtime/llm/client.go index ff85756..54c29b5 100644 --- a/runtime/llm/client.go +++ b/runtime/llm/client.go @@ -11,11 +11,15 @@ import ( "time" ) -// Client calls an OpenAI-compatible chat completions endpoint with streaming. +// Client calls a chat completions endpoint with streaming. +// 默认讲 OpenAI 兼容协议;Protocol 为 ProtocolAnthropic 时讲原生 +// Anthropic Messages(组包与 BuildBody 共用,见 anthropic.go)。 type Client struct { BaseURL string APIKey string HTTPClient *http.Client + // Protocol 空 = openai-chat(零值兼容所有直接构造点)。 + Protocol Protocol } // Stream implements StreamFunc. Transport and provider failures are encoded as @@ -24,6 +28,10 @@ func (c *Client) Stream(ctx context.Context, model string, messages []Message, t out := make(chan Event) go func() { defer close(out) + if c.Protocol == ProtocolAnthropic { + c.streamAnthropic(ctx, model, messages, tools, opts, out) + return + } c.run(ctx, model, messages, tools, opts, out) }() return out diff --git a/runtime/llm/providers.go b/runtime/llm/providers.go index 456f55a..10c9a31 100644 --- a/runtime/llm/providers.go +++ b/runtime/llm/providers.go @@ -140,7 +140,7 @@ type Provider struct { // Presets are OpenAI-compatible or native endpoints that only need an API key. func Presets() []Provider { return []Provider{ - {Name: "anthropic", Protocol: ProtocolOpenAIChat, BaseURL: "https://api.anthropic.com/v1", APIKeyEnv: "ANTHROPIC_API_KEY"}, + {Name: "anthropic", Protocol: ProtocolAnthropic, BaseURL: "https://api.anthropic.com", APIKeyEnv: "ANTHROPIC_API_KEY"}, {Name: "openai", Protocol: ProtocolOpenAIChat, BaseURL: "https://api.openai.com/v1", APIKeyEnv: "OPENAI_API_KEY"}, {Name: "openai-responses", Protocol: ProtocolOpenAIResponses, BaseURL: "https://api.openai.com/v1", APIKeyEnv: "OPENAI_API_KEY"}, {Name: "google", Protocol: ProtocolOpenAIChat, BaseURL: "https://generativelanguage.googleapis.com/v1beta/openai", APIKeyEnv: "GEMINI_API_KEY"}, @@ -196,12 +196,14 @@ func Lookup(name string, custom []Provider) (Provider, bool) { return Provider{}, false } -// ClientFor 返回会说该协议的客户端。responses、azure 以及原生 Anthropic/Google +// ClientFor 返回会说该协议的客户端。responses、azure 以及原生 Google // 协议当前没有接到客户端上,返回明确错误而不是发到错误的路径。 func ClientFor(p Provider) (*Client, error) { switch p.Protocol { case ProtocolOpenAIChat: return &Client{BaseURL: p.BaseURL, APIKey: p.APIKey}, nil + case ProtocolAnthropic: + return &Client{BaseURL: p.BaseURL, APIKey: p.APIKey, Protocol: ProtocolAnthropic}, nil default: return nil, fmt.Errorf("llm: protocol %q is not implemented", p.Protocol) } @@ -264,29 +266,10 @@ func BuildBody(protocol Protocol, model string, messages []Message, tools []Tool body["reasoning"] = map[string]any{"effort": effort} } return json.Marshal(body) - // ProtocolAnthropic 与 ProtocolGoogle 只供离线组包。 - // 当前 Client 只发 OpenAI chat completions,这两支没有接到客户端上。 - // 原生协议是后续阶段的事;预设表里的 anthropic / google 已改走兼容端点。 + // ProtocolGoogle 只供离线组包(客户端未接);Anthropic 与线上客户端 + // 共用 anthropicBody(同一组包,消灭离线/线上两张皮)。 case ProtocolAnthropic: - sys, msgs := anthropicMessages(messages) - body := map[string]any{ - "model": model, - "messages": msgs, - "max_tokens": opts.MaxTokens, - "stream": true, - } - // Anthropic 要求 max_tokens。0 表示调用方不限制:用足够大的默认值, - // 避免 4096 把长回复截断。真正的上限由调用方 StreamOptions.MaxTokens 决定。 - if body["max_tokens"] == 0 { - body["max_tokens"] = 128000 - } - if sys != "" { - body["system"] = sys - } - if budget := AnthropicThinkingBudget(opts.ThinkingLevel); budget > 0 { - body["thinking"] = map[string]any{"type": "enabled", "budget_tokens": budget} - } - return json.Marshal(body) + return anthropicBody(model, messages, tools, opts) case ProtocolGoogle: body := map[string]any{ "contents": googleContents(messages), @@ -317,40 +300,6 @@ func responsesInput(messages []Message) []map[string]any { return out } -func anthropicMessages(messages []Message) (string, []map[string]any) { - var system strings.Builder - var out []map[string]any - for _, m := range messages { - if m.Role == RoleSystem { - if system.Len() > 0 { - system.WriteString("\n") - } - system.WriteString(m.Content) - continue - } - role := "user" - if m.Role == RoleAssistant { - role = "assistant" - } - 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": role, "content": content}) - } - return system.String(), out -} - func googleContents(messages []Message) []map[string]any { var out []map[string]any for _, m := range messages { diff --git a/runtime/llm/providers_test.go b/runtime/llm/providers_test.go index eeb6c56..c30c3df 100644 --- a/runtime/llm/providers_test.go +++ b/runtime/llm/providers_test.go @@ -69,19 +69,29 @@ func TestModelsJSONAndBodies(t *testing.T) { } func TestPresetsAndClientForChatCompletions(t *testing.T) { - want := map[string]string{ - "anthropic": "https://api.anthropic.com/v1", - "google": "https://generativelanguage.googleapis.com/v1beta/openai", + // google 仍走 OpenAI 兼容端点;anthropic 自 N2 起是原生 Messages 协议。 + want := map[string]struct { + protocol Protocol + base string + }{ + "anthropic": {ProtocolAnthropic, "https://api.anthropic.com"}, + "google": {ProtocolOpenAIChat, "https://generativelanguage.googleapis.com/v1beta/openai"}, } for _, p := range Presets() { - base, ok := want[p.Name] + w, ok := want[p.Name] if !ok { continue } - if p.Protocol != ProtocolOpenAIChat || p.BaseURL != base { + if p.Protocol != w.protocol || p.BaseURL != w.base { t.Fatalf("%s preset: %+v", p.Name, p) } } + // 原生 anthropic 有真实客户端(审计 N4 关闭);responses/azure/google 仍 fail-closed。 + if p, ok := Lookup("anthropic", nil); !ok { + t.Fatal("anthropic preset missing") + } else if c, err := ClientFor(p); err != nil || c.Protocol != ProtocolAnthropic { + t.Fatalf("anthropic ClientFor: %v %+v", err, c) + } for _, name := range []string{"openai-responses", "azure"} { p, ok := Lookup(name, nil) if !ok { diff --git a/runtime/llm/testdata/anthropic/01-tool-use.sse b/runtime/llm/testdata/anthropic/01-tool-use.sse new file mode 100644 index 0000000..df0593a --- /dev/null +++ b/runtime/llm/testdata/anthropic/01-tool-use.sse @@ -0,0 +1,20 @@ +event: message_start +data: {"type":"message_start","message":{"id":"msg_01","type":"message","role":"assistant","model":"claude-fixture","usage":{"input_tokens":88,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}} + +event: content_block_start +data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_01ABC","name":"read_file"}} + +event: content_block_delta +data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"path\":"}} + +event: content_block_delta +data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"\"main.go\"}"}} + +event: content_block_stop +data: {"type":"content_block_stop","index":0} + +event: message_delta +data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":31}} + +event: message_stop +data: {"type":"message_stop"} diff --git a/runtime/llm/testdata/anthropic/02-thinking-text.sse b/runtime/llm/testdata/anthropic/02-thinking-text.sse new file mode 100644 index 0000000..1b19546 --- /dev/null +++ b/runtime/llm/testdata/anthropic/02-thinking-text.sse @@ -0,0 +1,29 @@ +event: message_start +data: {"type":"message_start","message":{"id":"msg_02","type":"message","role":"assistant","model":"claude-fixture","usage":{"input_tokens":142,"cache_read_input_tokens":60,"cache_creation_input_tokens":0}}} + +event: content_block_start +data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}} + +event: content_block_delta +data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"用户问 main.go,工具已返回。"}} + +event: content_block_delta +data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"EqoBCkgIARABGAIiQA"}} + +event: content_block_stop +data: {"type":"content_block_stop","index":0} + +event: content_block_start +data: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}} + +event: content_block_delta +data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"查到了,main.go 一共 42 行。"}} + +event: content_block_stop +data: {"type":"content_block_stop","index":1} + +event: message_delta +data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":57}} + +event: message_stop +data: {"type":"message_stop"} diff --git a/runtime/llm/testdata/anthropic/03-text.sse b/runtime/llm/testdata/anthropic/03-text.sse new file mode 100644 index 0000000..c2f9ea5 --- /dev/null +++ b/runtime/llm/testdata/anthropic/03-text.sse @@ -0,0 +1,17 @@ +event: message_start +data: {"type":"message_start","message":{"id":"msg_03","type":"message","role":"assistant","model":"claude-fixture","usage":{"input_tokens":201}}} + +event: content_block_start +data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + +event: content_block_delta +data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"42 行。"}} + +event: content_block_stop +data: {"type":"content_block_stop","index":0} + +event: message_delta +data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":9}} + +event: message_stop +data: {"type":"message_stop"} diff --git a/runtime/llm/types.go b/runtime/llm/types.go index 28f61e6..113bb8b 100644 --- a/runtime/llm/types.go +++ b/runtime/llm/types.go @@ -52,8 +52,21 @@ type Message struct { Images []Image `json:"images,omitempty"` ToolCallID string `json:"tool_call_id,omitempty"` ToolCalls []ToolCall `json:"tool_calls,omitempty"` - StopReason StopReason `json:"stop_reason,omitempty"` - IsError bool `json:"is_error,omitempty"` + // Thinking 是原生 Anthropic 路径的思考块(含签名),随 assistant 消息 + // 在请求里原样往返(Anthropic 要求思考块在内容块最前且签名不失)。 + // OpenAI 路径不写这个字段(reasoning 不落盘,见 sse.go 注释)。 + Thinking []ThinkingBlock `json:"thinking,omitempty"` + StopReason StopReason `json:"stop_reason,omitempty"` + IsError bool `json:"is_error,omitempty"` +} + +// ThinkingBlock 是一块 Anthropic 思考内容。Redacted 为真时 Data 是不透明 +// 密文(redacted_thinking),往返时原样回送。 +type ThinkingBlock struct { + Thinking string `json:"thinking,omitempty"` + Signature string `json:"signature,omitempty"` + Redacted bool `json:"redacted,omitempty"` + Data string `json:"data,omitempty"` } // ToolDecl is the JSON-schema declaration sent to the provider.