hunjing/shenshi/internal/transform/anthropic.go

// Anthropic 出站适配器：IR → Anthropic Messages API（SSE 事件流）。
package transform

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"time"

	"hunjing/shenshi/internal/ir"
)

// AnthropicAdaptor Anthropic Messages 端点适配器。
type AnthropicAdaptor struct {
	EndpointName  string
	BaseURL       string
	APIKey        string
	UpstreamModel string
	Client        *http.Client
}

// Name 实现 Adaptor。
func (a *AnthropicAdaptor) Name() string { return a.EndpointName }

type antMessage struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

// 可选标量一律指针 + omitempty（shenshi-gateway.md §5 DTO 纪律）。
type antRequest struct {
	Model       string       `json:"model"`
	Messages    []antMessage `json:"messages"`
	MaxTokens   *int         `json:"max_tokens,omitempty"`
	Temperature *float64     `json:"temperature,omitempty"`
	Stream      bool         `json:"stream"`
	System      *string      `json:"system,omitempty"`
}

// antEvent Anthropic SSE 事件：message_start 的用量在 message.usage，
// message_delta 的用量在顶层 usage（两处形态不同，指针区分可选标量）。
type antEvent struct {
	Type  string `json:"type"`
	Delta struct {
		Type string `json:"type"`
		Text string `json:"text"`
	} `json:"delta"`
	Message *struct {
		Usage *struct {
			InputTokens  int `json:"input_tokens"`
			OutputTokens int `json:"output_tokens"`
		} `json:"usage"`
	} `json:"message"`
	Usage *struct {
		OutputTokens int `json:"output_tokens"`
	} `json:"usage"`
}

// Stream 实现 Adaptor：Anthropic SSE（message_delta / content_block_delta 等）转 IR Delta。
func (a *AnthropicAdaptor) Stream(ctx context.Context, req *ir.ChatRequest, out chan<- ir.Delta) error {
	body := antRequest{
		Model:       a.UpstreamModel,
		Messages:    make([]antMessage, 0, len(req.Messages)),
		MaxTokens:   req.MaxTokens,
		Temperature: req.Temperature,
		Stream:      true,
	}
	// Anthropic 的 system 是顶层字段而非消息序列首条
	var system strings.Builder
	for _, m := range req.Messages {
		if m.Role == "system" {
			if system.Len() > 0 {
				system.WriteString("\n\n")
			}
			system.WriteString(m.Content)
			continue
		}
		role := m.Role
		if role == "tool" {
			role = "user" // 最小映射：tool 结果按用户侧内容回传（工具循环随百工 M5 完整化）
		}
		body.Messages = append(body.Messages, antMessage{Role: role, Content: m.Content})
	}
	if system.Len() > 0 {
		s := system.String()
		body.System = &s
	}
	if body.MaxTokens == nil {
		def := 4096 // Anthropic 要求 max_tokens 必填
		body.MaxTokens = &def
	}

	payload, err := json.Marshal(body)
	if err != nil {
		return err
	}
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost,
		strings.TrimRight(a.BaseURL, "/")+"/v1/messages", bytes.NewReader(payload))
	if err != nil {
		return err
	}
	httpReq.Header.Set("content-type", "application/json")
	httpReq.Header.Set("x-api-key", a.APIKey)
	httpReq.Header.Set("anthropic-version", "2023-06-01")

	client := a.Client
	if client == nil {
		client = &http.Client{Timeout: 120 * time.Second}
	}
	resp, err := client.Do(httpReq)
	if err != nil {
		return &UpstreamError{StatusCode: 0, Message: err.Error()}
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode >= 400 {
		buf := new(bytes.Buffer)
		_, _ = buf.ReadFrom(resp.Body)
		return &UpstreamError{StatusCode: resp.StatusCode, Message: buf.String()}
	}

	var finish string
	var usagePrompt, usageCompletion int
	scanner := bufio.NewScanner(resp.Body)
	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
	for scanner.Scan() {
		line := scanner.Text()
		if !strings.HasPrefix(line, "data:") {
			continue
		}
		data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
		if data == "" {
			continue
		}
		var ev antEvent
		if err := json.Unmarshal([]byte(data), &ev); err != nil {
			continue
		}
		switch ev.Type {
		case "content_block_delta":
			if ev.Delta.Type == "text_delta" && ev.Delta.Text != "" {
				out <- ir.Delta{Delta: ev.Delta.Text}
			}
		case "message_delta":
			// 结束与用量在 message_delta 顶层 usage 汇总（output_tokens）
			if ev.Usage != nil {
				usageCompletion = ev.Usage.OutputTokens
			}
			finish = "stop"
		case "message_start":
			if ev.Message != nil && ev.Message.Usage != nil {
				usagePrompt = ev.Message.Usage.InputTokens
			}
		case "error":
			return fmt.Errorf("上游流内错误: %s", data)
		}
	}
	if err := scanner.Err(); err != nil {
		return err
	}
	if finish != "" {
		out <- ir.Delta{FinishReason: finish, UsagePromptTokens: usagePrompt, UsageCompletionTokens: usageCompletion}
	}
	return nil
}