Sakurasan
d5dac5a6fc
llm（三协议 Provider）/ session（JSONL 持久化）/ tools（Registry、围栏、 builtin、computer）/ cdp（browser）/ skills / mcp / agent（循环、两级压缩、 编排、审批门控）+ 底座 jsonl/textx/execx/diag/scratch/spill/imagecache。 产品装配代码（config 桥接、schedule 工具、memory/plan/retro/prompt） 留在 AutoGo，宿主接入方式见 MIGRATION.md。
355 lines
11 KiB
Go
355 lines
11 KiB
Go
package llm
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strings"
|
||
)
|
||
|
||
// ResponsesProvider 走 OpenAI Responses API（/responses，SSE 流式）。
|
||
// 与 Chat Completions 的差异：事件驱动流（response.output_text.delta 等）、
|
||
// 工具调用经 function_call/function_call_output 消息往返、图片用 input_image。
|
||
type ResponsesProvider struct {
|
||
BaseURL string
|
||
APIKey string
|
||
Model string
|
||
HTTP *http.Client
|
||
}
|
||
|
||
// SetModel 运行时切换模型（ModelSwitcher）。
|
||
func (p *ResponsesProvider) SetModel(model string) {
|
||
p.Model = model
|
||
}
|
||
|
||
// CurrentModel 报告当前模型（CurrentModeler）。
|
||
func (p *ResponsesProvider) CurrentModel() string { return p.Model }
|
||
|
||
// Identity 自报协议与当前模型（Identifier）。
|
||
func (p *ResponsesProvider) Identity() (string, string) { return FormatResponses, p.Model }
|
||
|
||
// ListModels 枚举网关可用模型（ModelLister），与 Chat Completions 同一 GET /models。
|
||
func (p *ResponsesProvider) ListModels(ctx context.Context) ([]string, error) {
|
||
op := &OpenAIProvider{BaseURL: p.BaseURL, APIKey: p.APIKey, HTTP: p.HTTP}
|
||
return op.ListModels(ctx)
|
||
}
|
||
|
||
// rsEvent 是 /responses SSE 事件包络（外层只取 type，其余按事件类型二次解码）。
|
||
type rsEvent struct {
|
||
Type string `json:"type"`
|
||
}
|
||
|
||
// rsItemEvent 载荷：response.output_item.added / .done 的 function_call 项。
|
||
type rsItemEvent struct {
|
||
Item *rsOutputItem `json:"item"`
|
||
}
|
||
|
||
// rsArgsDeltaEvent 载荷：response.function_call_arguments.delta。
|
||
type rsArgsDeltaEvent struct {
|
||
ItemID string `json:"item_id"`
|
||
CallID string `json:"call_id"`
|
||
Arguments string `json:"delta"`
|
||
}
|
||
|
||
// rsCompletedEvent 载荷：response.completed / response.incomplete。
|
||
type rsCompletedEvent struct {
|
||
Response struct {
|
||
Output []rsOutputItem `json:"output"`
|
||
Usage rsUsageWrap `json:"usage"`
|
||
// Status 是 completed / incomplete；incomplete 时必须看 IncompleteDetails
|
||
// 才知道是不是"输出上限截住"（另一种是内容审核拦下）。
|
||
Status string `json:"status"`
|
||
IncompleteDetails struct {
|
||
Reason string `json:"reason"`
|
||
} `json:"incomplete_details"`
|
||
} `json:"response"`
|
||
}
|
||
|
||
type rsOutputItem struct {
|
||
Type string `json:"type"`
|
||
ID string `json:"id"`
|
||
CallID string `json:"call_id"`
|
||
Name string `json:"name"`
|
||
Arguments string `json:"arguments"`
|
||
}
|
||
|
||
type rsUsageWrap struct {
|
||
InputTokens int `json:"input_tokens"`
|
||
OutputTokens int `json:"output_tokens"`
|
||
// 明细：input_tokens 已含缓存命中部分（与 Chat Completions 同口径）。
|
||
InputDetails struct {
|
||
Cached int `json:"cached_tokens"`
|
||
} `json:"input_tokens_details"`
|
||
OutputDetails struct {
|
||
Reasoning int `json:"reasoning_tokens"`
|
||
} `json:"output_tokens_details"`
|
||
}
|
||
|
||
func (p *ResponsesProvider) Stream(ctx context.Context, system string, msgs []Message, tools []ToolDef, opts StreamOptions, onDelta func(StreamDelta)) (Message, error) {
|
||
body := map[string]any{
|
||
"model": p.Model,
|
||
"input": rsBuildInput(joinSystem(system, opts.SystemDynamic), msgs),
|
||
"stream": true,
|
||
"store": false,
|
||
"max_output_tokens": 8192,
|
||
}
|
||
if len(tools) > 0 {
|
||
body["tools"] = rsBuildTools(tools)
|
||
if opts.DisableTools {
|
||
body["tool_choice"] = "none"
|
||
}
|
||
}
|
||
// 思考级别：仅开启时发送；推理 token 计入 max_output_tokens，故一并抬高上限防截断。
|
||
if opts.Thinking.Enabled() {
|
||
body["reasoning"] = map[string]any{"effort": string(opts.Thinking)}
|
||
body["max_output_tokens"] = 16384
|
||
}
|
||
payload, err := json.Marshal(body)
|
||
if err != nil {
|
||
return Message{}, err
|
||
}
|
||
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
|
||
strings.TrimRight(p.BaseURL, "/")+"/responses", bytes.NewReader(payload))
|
||
if err != nil {
|
||
return Message{}, err
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("Authorization", "Bearer "+p.APIKey)
|
||
|
||
resp, err := doWithRetry(ctx, p.http(), req, payload)
|
||
if err != nil {
|
||
return Message{}, err
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode != http.StatusOK {
|
||
var buf bytes.Buffer
|
||
_, _ = buf.ReadFrom(io.LimitReader(resp.Body, 4096))
|
||
return Message{}, fmt.Errorf("responses: HTTP %d: %s", resp.StatusCode, buf.String())
|
||
}
|
||
|
||
final := Message{Role: RoleAssistant}
|
||
var st streamState
|
||
// callID -> 增量参数缓冲（function_call_arguments.delta 事件流）。
|
||
argBuf := map[string]*ToolCall{}
|
||
// itemID -> callID（function_call_arguments 事件只带 item_id/或 call_id，二者取其一）。
|
||
itemToCall := map[string]string{}
|
||
emitted := map[string]bool{}
|
||
emitCall := func(tc ToolCall) {
|
||
if onDelta == nil || tc.ID == "" || emitted[tc.ID] {
|
||
return
|
||
}
|
||
emitted[tc.ID] = true
|
||
c := tc
|
||
onDelta(StreamDelta{ToolCall: &c})
|
||
}
|
||
|
||
sg := guardStream(resp.Body, streamIdleTimeout)
|
||
defer sg.Stop()
|
||
sc := bufio.NewScanner(sg)
|
||
sc.Buffer(make([]byte, 1<<20), 1<<20)
|
||
for sc.Scan() {
|
||
line := sc.Text()
|
||
if !strings.HasPrefix(line, "data: ") {
|
||
continue
|
||
}
|
||
raw := []byte(strings.TrimPrefix(line, "data: "))
|
||
var env rsEvent
|
||
if err := json.Unmarshal(raw, &env); err != nil {
|
||
continue
|
||
}
|
||
switch env.Type {
|
||
case "response.output_item.added", "response.output_item.done":
|
||
var ev rsItemEvent
|
||
if err := json.Unmarshal(raw, &ev); err != nil || ev.Item == nil || ev.Item.Type != "function_call" {
|
||
continue
|
||
}
|
||
// added 通常只带名字，done 带完整参数；参数为空时保留增量缓冲里已收到的那份。
|
||
tc := argBuf[ev.Item.CallID]
|
||
if tc == nil {
|
||
tc = &ToolCall{ID: ev.Item.CallID}
|
||
argBuf[ev.Item.CallID] = tc
|
||
}
|
||
if ev.Item.Name != "" {
|
||
tc.Name = ev.Item.Name
|
||
}
|
||
if ev.Item.Arguments != "" {
|
||
tc.Input = ev.Item.Arguments
|
||
}
|
||
if ev.Item.ID != "" {
|
||
itemToCall[ev.Item.ID] = ev.Item.CallID
|
||
}
|
||
// done 事件的 parameters 已完整：提前通知，只读工具可先跑。
|
||
if env.Type == "response.output_item.done" {
|
||
emitCall(*tc)
|
||
}
|
||
case "response.function_call_arguments.delta":
|
||
var ev rsArgsDeltaEvent
|
||
if err := json.Unmarshal(raw, &ev); err != nil {
|
||
continue
|
||
}
|
||
callID := ev.CallID
|
||
if callID == "" {
|
||
callID = itemToCall[ev.ItemID] // 某些网关只带 item_id
|
||
}
|
||
if tc := argBuf[callID]; tc != nil {
|
||
tc.Input += ev.Arguments
|
||
}
|
||
// 思考内容：summary（多网关可见）与完整推理文本都收。
|
||
case "response.reasoning_summary_text.delta", "response.reasoning_text.delta":
|
||
var ev struct {
|
||
Delta string `json:"delta"`
|
||
}
|
||
if err := json.Unmarshal(raw, &ev); err != nil || ev.Delta == "" {
|
||
continue
|
||
}
|
||
final.Reasoning += ev.Delta
|
||
if onDelta != nil {
|
||
onDelta(StreamDelta{Reasoning: ev.Delta})
|
||
}
|
||
case "response.output_text.delta":
|
||
var ev struct {
|
||
Delta string `json:"delta"`
|
||
}
|
||
if err := json.Unmarshal(raw, &ev); err != nil || ev.Delta == "" {
|
||
continue
|
||
}
|
||
final.Content += ev.Delta
|
||
if onDelta != nil {
|
||
onDelta(StreamDelta{Text: ev.Delta})
|
||
}
|
||
case "response.completed", "response.incomplete":
|
||
var ev rsCompletedEvent
|
||
if err := json.Unmarshal(raw, &ev); err == nil {
|
||
final.Usage = &Usage{
|
||
InputTokens: ev.Response.Usage.InputTokens,
|
||
OutputTokens: ev.Response.Usage.OutputTokens,
|
||
CacheReadTokens: ev.Response.Usage.InputDetails.Cached,
|
||
ReasoningTokens: ev.Response.Usage.OutputDetails.Reasoning,
|
||
}
|
||
// 以 completed 事件的权威 output 为准重建 tool calls（增量网关兜底）。
|
||
final.ToolCalls = nil
|
||
for _, it := range ev.Response.Output {
|
||
if it.Type == "function_call" && it.Name != "" {
|
||
tc := ToolCall{ID: it.CallID, Name: it.Name, Input: it.Arguments}
|
||
final.ToolCalls = append(final.ToolCalls, tc)
|
||
emitCall(tc)
|
||
}
|
||
}
|
||
// incomplete 有两种：输出上限截住（可续写）与内容审核拦下（续写无意义）。
|
||
if ev.Response.Status == "incomplete" && marksOutputLimit(ev.Response.IncompleteDetails.Reason) {
|
||
final.OutputLimited = true
|
||
}
|
||
}
|
||
st.doneSeen = true
|
||
return completion("responses", final, st)
|
||
}
|
||
}
|
||
if err := sc.Err(); err != nil {
|
||
if stall := sg.Err(); stall != nil {
|
||
return Message{}, fmt.Errorf("responses: %w", stall)
|
||
}
|
||
return Message{}, fmt.Errorf("responses: stream: %w", err)
|
||
}
|
||
// 看门狗掐断的流不走"尽力拼装"：那会把半截输出当成一轮完整回复。
|
||
if stall := sg.Err(); stall != nil {
|
||
return Message{}, fmt.Errorf("responses: %w", stall)
|
||
}
|
||
// 无 response.completed（连接截断）：用增量缓冲尽力拼装。
|
||
for _, tc := range argBuf {
|
||
if tc.Name != "" {
|
||
final.ToolCalls = append(final.ToolCalls, *tc)
|
||
emitCall(*tc)
|
||
}
|
||
}
|
||
return completion("responses", final, st)
|
||
}
|
||
|
||
// rsBuildInput 构造 /responses 的 input 数组：system 作为首条 system 消息，
|
||
// tool 结果映射为 function_call_output，assistant 工具调用拆成 function_call 条目。
|
||
func rsBuildInput(system string, msgs []Message) []map[string]any {
|
||
out := []map[string]any{}
|
||
if system != "" {
|
||
out = append(out, map[string]any{"role": "system", "content": system})
|
||
}
|
||
for _, m := range msgs {
|
||
switch {
|
||
case m.Role == RoleTool:
|
||
out = append(out, map[string]any{
|
||
"type": "function_call_output",
|
||
"call_id": m.ToolCallID,
|
||
"output": m.Content,
|
||
})
|
||
if len(m.Images) > 0 {
|
||
// function_call_output 不收图片：附一条 user 消息带图。
|
||
out = append(out, map[string]any{
|
||
"role": "user",
|
||
"content": rsContentParts("[image(s) attached from tool result]", m.Images),
|
||
})
|
||
}
|
||
case len(m.ToolCalls) > 0:
|
||
// assistant 的 function_call 拆成独立条目；文本部分单独一条。
|
||
if m.Content != "" {
|
||
out = append(out, map[string]any{"role": "assistant", "content": m.Content})
|
||
}
|
||
for _, tc := range m.ToolCalls {
|
||
out = append(out, map[string]any{
|
||
"type": "function_call",
|
||
"call_id": tc.ID,
|
||
"name": tc.Name,
|
||
"arguments": tc.Input,
|
||
})
|
||
}
|
||
case len(m.Images) > 0:
|
||
out = append(out, map[string]any{"role": string(m.Role), "content": rsContentParts(m.Content, m.Images)})
|
||
default:
|
||
out = append(out, map[string]any{"role": string(m.Role), "content": m.Content})
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// rsContentParts 构造多模态 content（text + input_image）。
|
||
func rsContentParts(text string, imgs []Image) []map[string]any {
|
||
parts := []map[string]any{}
|
||
if text != "" {
|
||
parts = append(parts, map[string]any{"type": "input_text", "text": text})
|
||
}
|
||
for _, img := range imgs {
|
||
b64, mt, err := loadImageRef(img)
|
||
if err != nil {
|
||
parts = append(parts, map[string]any{"type": "input_text", "text": imageUnavailable(img, err)})
|
||
continue
|
||
}
|
||
parts = append(parts, map[string]any{
|
||
"type": "input_image",
|
||
"image_url": "data:" + mt + ";base64," + b64,
|
||
})
|
||
}
|
||
return parts
|
||
}
|
||
|
||
func rsBuildTools(tools []ToolDef) []map[string]any {
|
||
out := make([]map[string]any, 0, len(tools))
|
||
for _, t := range tools {
|
||
out = append(out, map[string]any{
|
||
"type": "function",
|
||
"name": t.Name,
|
||
"description": t.Description,
|
||
"parameters": t.Parameters,
|
||
})
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (p *ResponsesProvider) http() *http.Client {
|
||
if p.HTTP != nil {
|
||
return p.HTTP
|
||
}
|
||
return defaultHTTPClient
|
||
}
|