toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(openai): provider tool call을 그대로 전달한다

Browse source
This commit is contained in:
toki 2026-06-27 21:44:49 +09:00
parent 8a17e3a057
commit 7dc0788f88
14 changed files with 736 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
agent-contract/inner/edge-node-runtime-wire.md
View file

@ -44,6 +44,7 @@ Edge는 Node 연결을 수락하고, Node는 연결 직후 등록 요청을 보
- `RunRequest.input`: adapter가 해석할 실행 입력이다. CLI 실행에서는 prompt 계열 입력으로 변환된다.
- `RunRequest.metadata`: caller-defined 실행 metadata다. workspace 자체는 별도 `workspace` 필드로 전달한다.
- `RunEvent.type`: `start`, `delta`, `complete`, `error`, `cancelled` 같은 실행 이벤트 종류다.
- `RunEvent.metadata["openai_tool_calls"]`: OpenAI-compatible provider adapter가 native `tool_calls`를 반환했을 때 완료 이벤트에 싣는 JSON 배열이다. Edge OpenAI-compatible 표면은 이 값을 `message.tool_calls` 또는 stream `delta.tool_calls`로 복원한다. provider assistant content 텍스트를 이 값으로 파싱/합성하지 않는다.
- `NodeCommandRequest.type`: 실행이 아닌 조회/제어성 명령이다. adapter execution 요청과 섞지 않는다.
- `NodeConfigPayload.adapters`: Edge가 Node에 내려주는 adapter instance 설정이다.
- `AdapterConfig.name`: node 내부 stable adapter instance identity다. 비어 있으면 legacy single-instance type 이름과 동등하다.

18
agent-contract/outer/openai-compatible-api.md
View file

@ -165,17 +165,13 @@ Workspace-bound route는 workspace가 없거나 상대 경로이면 OpenAI-compa
- `parallel_tool_calls`
- `stream_options`
`tools`가 있는 Chat Completions 요청에서 Edge는 backend-native tool call stream을 일반 구조화 이벤트로 중계하지 않는다.
단, backend가 assistant content에 Cline-style 텍스트 tool call 블록을 출력하고 그 function name이 요청의 `tools[].function.name`에 포함되어 있으면 Edge는 해당 블록을 OpenAI-compatible `message.tool_calls` 또는 stream `delta.tool_calls`로 변환하고 `finish_reason: "tool_calls"`를 반환한다.
지원하는 텍스트 블록의 최소 형태는 `<tool_call><function=<name>><parameter=<key>>JSON-or-text</parameter></function></tool_call>` 또는 `{{function_name(key=Python/JSON-like-literal)}}`이다.
템플릿 호출 형태의 인자는 single-quoted string, `True`/`False`/`None`, 배열, 객체를 JSON `arguments` 문자열로 정규화한다.
텍스트 tool call을 구조화할 때 Edge는 요청의 `tools[].function.parameters` schema를 기준으로 arguments를 정규화한다.
예를 들어 tool schema가 `commands: string[]`만 허용하면 command 객체 입력도 shell string 배열로 접고, `commands: {command,args}[]`를 허용하면 shell 문법이 없는 명령을 structured argv로 만든다.
tool schema가 top-level `command`/`cmd` string만 허용하면 `commands` 배열의 첫 명령을 해당 field로 접는다.
schema에 없는 UI 설명용 `description`이나 실행 위치 힌트용 `runInTerminal`은 command 객체와 최상위 args에서 제거한다.
단, `cd`, `command -v`, `&&`, pipe, redirect, quote 등 shell 해석이 필요한 명령은 schema가 허용할 때 `commands: ["cd /work && git status"]` 같은 shell string으로 유지한다.
`{command:"cd", args:["/work","&&","git","status"]}`처럼 shell builtin이나 shell operator가 structured argv에 섞인 형태도 direct-exec로 실행되지 않도록 shell string으로 정규화한다.
호환성을 위해 `tool_choice`가 생략되었거나 `auto`/강제 tool 선택 형태로 들어오면 내부 실행 입력에서는 `tool_choice: "none"`으로 낮춰 백엔드의 auto tool-calling 요구 조건에 의해 요청 전체가 실패하지 않게 한다.
`tools`가 있는 Chat Completions 요청에서 provider route(`openai_compat`, `vllm`, `ollama`, provider pool)는 `tool_choice`를 클라이언트 요청값 그대로 backend에 전달한다. 클라이언트가 `tool_choice`를 생략하면 Edge도 생략한다.
provider가 native OpenAI-compatible `tool_calls`를 반환하면 Node는 내부 `RunEvent.metadata["openai_tool_calls"]` JSON으로 보존하고, Edge는 이를 OpenAI-compatible `message.tool_calls` 또는 stream `delta.tool_calls`로 반환하며 `finish_reason: "tool_calls"`를 사용한다.
provider route에서 assistant content에 `<tool_call>`, `{{function(...)}}`, `[Calling tool: ...]` 같은 텍스트가 들어와도 Edge는 이를 파싱하거나 OpenAI `tool_calls`로 합성하지 않는다. 해당 텍스트는 backend가 반환한 content로 그대로 둔다.
CLI route(`adapter: "cli"`)는 native backend tool calling이 없으므로, 호환 fallback으로만 assistant content의 Cline-style 텍스트 tool call 블록을 OpenAI-compatible `tool_calls`로 합성할 수 있다. 이 fallback이 지원하는 텍스트 블록의 최소 형태는 `<tool_call><function=<name>><parameter=<key>>JSON-or-text</parameter></function></tool_call>` 또는 `{{function_name(key=Python/JSON-like-literal)}}`이다.
CLI fallback에서 텍스트 tool call을 구조화할 때만 Edge는 요청의 `tools[].function.parameters` schema를 기준으로 arguments를 정규화한다. 예를 들어 tool schema가 `commands: string[]`만 허용하면 command 객체 입력도 shell string 배열로 접고, `commands: {command,args}[]`를 허용하면 shell 문법이 없는 명령을 structured argv로 만든다.
CLI fallback에서 schema에 없는 UI 설명용 `description`이나 실행 위치 힌트용 `runInTerminal`은 command 객체와 최상위 args에서 제거한다. 단, `cd`, `command -v`, `&&`, pipe, redirect, quote 등 shell 해석이 필요한 명령은 schema가 허용할 때 `commands: ["cd /work && git status"]` 같은 shell string으로 유지한다.
CLI route에서는 backend auto tool-calling 요구 조건으로 요청이 실패하지 않도록 내부 실행 입력의 `tool_choice`를 `"none"`으로 낮춘다.
`parallel_tool_calls`와 `stream_options`는 클라이언트 호환성을 위해 수신하지만 현재 Edge 실행 의미에는 반영하지 않는다.
금지:

34
apps/edge/internal/openai/chat_handler.go
View file

@ -54,7 +54,7 @@ func (s *Server) handleChatCompletions(w http.ResponseWriter, r *http.Request) {
messages = prependSystemMessage(messages, instruction)
}
prompt := promptFromMessages(messages)
input := req.runInput(prompt, messages, outputPolicy.Strict)
input := req.runInput(prompt, messages, outputPolicy.Strict, routeSupportsNativeToolCalls(dispatch))
s.logger.Info("openai chat completion input",
zap.String("model", req.Model),
zap.String("target", dispatch.Target),
@ -142,19 +142,41 @@ func chatRunMetadata(runMeta map[string]string, req chatCompletionRequest, outpu
return runMeta
}
func routeSupportsNativeToolCalls(dispatch routeDispatch) bool {
if dispatch.ProviderPool {
return true
}
switch strings.TrimSpace(dispatch.Adapter) {
case "openai_compat", "vllm", "ollama":
return true
default:
return false
}
}
func shouldSynthesizeTextToolCalls(dispatch edgeservice.RunDispatch) bool {
return strings.TrimSpace(dispatch.Adapter) == "cli"
}
func (s *Server) completeChatCompletion(w http.ResponseWriter, r *http.Request, req chatCompletionRequest, handle edgeservice.RunResult, outputPolicy strictOutputPolicy) {
text, reasoning, finishReason, usage, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
text, reasoning, finishReason, nativeToolCalls, usage, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
if err != nil {
writeError(w, httpStatusForRunError(err), "run_error", err.Error())
return
}
text, reasoning, normalized := normalizeCompletionOutput(outputPolicy, text, reasoning)
message := chatMessage{Role: "assistant", Content: text, ReasoningContent: reasoning}
if cleaned, toolCalls := synthesizeToolCallsFromText(text, req.Tools, handle.Dispatch().RunID); len(toolCalls) > 0 {
message.Content = cleaned
message.ReasoningContent = ""
message.ToolCalls = toolCalls
if len(nativeToolCalls) > 0 {
message.ToolCalls = nativeToolCalls
finishReason = "tool_calls"
} else if shouldSynthesizeTextToolCalls(handle.Dispatch()) {
cleaned, toolCalls := synthesizeToolCallsFromText(text, req.Tools, handle.Dispatch().RunID)
if len(toolCalls) > 0 {
message.Content = cleaned
message.ReasoningContent = ""
message.ToolCalls = toolCalls
finishReason = "tool_calls"
}
}
s.logger.Info("openai chat completion output",
zap.String("run_id", handle.Dispatch().RunID),

2
apps/edge/internal/openai/responses_handler.go
View file

@ -140,7 +140,7 @@ func decodeResponsesRequest(dec *json.Decoder, req *responsesRequest) error {
}
func (s *Server) completeResponse(w http.ResponseWriter, r *http.Request, req responsesRequest, handle edgeservice.RunResult, outputPolicy strictOutputPolicy) {
text, reasoning, _, usage, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
text, reasoning, _, _, usage, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
if err != nil {
writeError(w, httpStatusForRunError(err), "run_error", err.Error())
return

38
apps/edge/internal/openai/run_result.go
View file

@ -2,6 +2,7 @@ package openai
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
@ -9,9 +10,11 @@ import (
edgeservice "iop/apps/edge/internal/service"
)
func collectRunResult(ctx context.Context, stream edgeservice.RunStream, timeout time.Duration) (string, string, string, *openAIUsage, error) {
const runtimeMetadataOpenAIToolCalls = "openai_tool_calls"
func collectRunResult(ctx context.Context, stream edgeservice.RunStream, timeout time.Duration) (string, string, string, []any, *openAIUsage, error) {
if stream.Events == nil {
return "", "", "", nil, fmt.Errorf("run stream unavailable")
return "", "", "", nil, nil, fmt.Errorf("run stream unavailable")
}
var contentBuilder strings.Builder
var reasoningBuilder strings.Builder
@ -22,20 +25,20 @@ func collectRunResult(ctx context.Context, stream edgeservice.RunStream, timeout
for {
select {
case <-ctx.Done():
return "", "", "", nil, ctx.Err()
return "", "", "", nil, nil, ctx.Err()
case <-timer.C:
return "", "", "", nil, fmt.Errorf("run timed out")
return "", "", "", nil, nil, fmt.Errorf("run timed out")
case nodeEvent, ok := <-stream.NodeEvents:
if !ok {
stream.NodeEvents = nil
continue
}
if edgeservice.IsNodeDisconnected(nodeEvent) {
return "", "", "", nil, fmt.Errorf("node disconnected")
return "", "", "", nil, nil, fmt.Errorf("node disconnected")
}
case event, ok := <-stream.Events:
if !ok {
return "", "", "", nil, fmt.Errorf("run stream closed")
return "", "", "", nil, nil, fmt.Errorf("run stream closed")
}
if event == nil {
continue
@ -49,6 +52,10 @@ func collectRunResult(ctx context.Context, stream edgeservice.RunStream, timeout
if reason := event.GetMetadata()["finish_reason"]; reason != "" {
finishReason = reason
}
toolCalls := toolCallsFromRunMetadata(event.GetMetadata())
if len(toolCalls) > 0 {
finishReason = "tool_calls"
}
if u := event.GetUsage(); u != nil {
usage = &openAIUsage{
PromptTokens: int(u.GetInputTokens()),
@ -56,7 +63,7 @@ func collectRunResult(ctx context.Context, stream edgeservice.RunStream, timeout
TotalTokens: int(u.GetInputTokens() + u.GetOutputTokens()),
}
}
return contentBuilder.String(), reasoningBuilder.String(), finishReason, usage, nil
return contentBuilder.String(), reasoningBuilder.String(), finishReason, toolCalls, usage, nil
case "error", "cancelled":
msg := event.GetError()
if msg == "" {
@ -65,8 +72,23 @@ func collectRunResult(ctx context.Context, stream edgeservice.RunStream, timeout
if msg == "" {
msg = "run failed"
}
return "", "", "", nil, fmt.Errorf("%s", msg)
return "", "", "", nil, nil, fmt.Errorf("%s", msg)
}
}
}
}
func toolCallsFromRunMetadata(metadata map[string]string) []any {
raw := strings.TrimSpace(metadata[runtimeMetadataOpenAIToolCalls])
if raw == "" {
return nil
}
var toolCalls []any
if err := json.Unmarshal([]byte(raw), &toolCalls); err != nil {
return nil
}
if len(toolCalls) == 0 {
return nil
}
return toolCalls
}

159
apps/edge/internal/openai/server_test.go
View file

@ -70,6 +70,7 @@ func (s *fakeRunService) SubmitRun(_ context.Context, req edgeservice.SubmitRunR
RunDispatch: edgeservice.RunDispatch{
RunID: "run-test",
ModelGroupKey: req.ModelGroupKey,
Adapter: req.Adapter,
Target: req.Target,
TimeoutSec: 5,
},
@ -226,12 +227,12 @@ func TestChatCompletionsPassesStandardOptions(t *testing.T) {
if tools, ok := fake.req.Input["tools"].([]any); !ok || len(tools) != 1 {
t.Fatalf("tools not passed: %+v", fake.req.Input["tools"])
}
if fake.req.Input["tool_choice"] != "none" {
t.Fatalf("tool_choice not downgraded: %+v", fake.req.Input["tool_choice"])
if _, ok := fake.req.Input["tool_choice"]; ok {
t.Fatalf("tool_choice must be omitted when the client omits it: %+v", fake.req.Input["tool_choice"])
}
}
func TestChatCompletionsDowngradesToolChoiceAutoToNone(t *testing.T) {
func TestChatCompletionsPassesProviderToolChoice(t *testing.T) {
fake := &fakeRunService{events: make(chan *iop.RunEvent, 2)}
fake.events <- &iop.RunEvent{Type: "delta", Delta: "ok"}
fake.events <- &iop.RunEvent{Type: "complete"}
@ -252,8 +253,8 @@ func TestChatCompletionsDowngradesToolChoiceAutoToNone(t *testing.T) {
if w.Code != http.StatusOK {
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
}
if fake.req.Input["tool_choice"] != "none" {
t.Fatalf("tool_choice: got %+v, want none", fake.req.Input["tool_choice"])
if fake.req.Input["tool_choice"] != "auto" {
t.Fatalf("tool_choice: got %+v, want auto", fake.req.Input["tool_choice"])
}
if _, ok := fake.req.Input["parallel_tool_calls"]; ok {
t.Fatalf("parallel_tool_calls must not be forwarded: %+v", fake.req.Input)
@ -263,12 +264,36 @@ func TestChatCompletionsDowngradesToolChoiceAutoToNone(t *testing.T) {
}
}
func TestChatCompletionsDowngradesCLIToolChoiceAutoToNone(t *testing.T) {
fake := &fakeRunService{events: make(chan *iop.RunEvent, 2)}
fake.events <- &iop.RunEvent{Type: "delta", Delta: "ok"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "cli", Target: "codex"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"codex-cli",
"messages":[{"role":"user","content":"hi"}],
"tools":[{"type":"function","function":{"name":"lookup"}}],
"tool_choice":"auto"
}`))
w := httptest.NewRecorder()
srv.handleChatCompletions(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
}
if fake.req.Input["tool_choice"] != "none" {
t.Fatalf("tool_choice: got %+v, want none", fake.req.Input["tool_choice"])
}
}
func TestChatCompletionsSynthesizesToolCallsFromClineTextBlock(t *testing.T) {
fake := &fakeRunService{events: make(chan *iop.RunEvent, 2)}
fake.events <- &iop.RunEvent{Type: "delta", Delta: "먼저 현재 git 상태를 확인하겠습니다.\n\n<tool_call>\n<function=run_commands>\n<parameter=commands>\n[\"cd /config/workspace/iop && git status\"]\n</parameter>\n</function>\n</tool_call>"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
srv := NewServer(config.EdgeOpenAIConf{Adapter: "cli", Target: "codex"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"qwen3.6:35b",
"messages":[{"role":"user","content":"status"}],
@ -327,7 +352,7 @@ func TestChatCompletionsSynthesizesToolCallsFromClineTemplateCall(t *testing.T)
fake.events <- &iop.RunEvent{Type: "delta", Delta: "사용자가 변경 내용을 푸시해달라고 요청하셨습니다. 먼저 현재 git 상태를 확인하여 변경된 파일과 커밋 가능한 내용을 파악하겠습니다.\n\n{{run_commands(commands=[{'command': 'git status', 'description': '현재 git 상태 확인', 'runInTerminal': True}]})}}"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
srv := NewServer(config.EdgeOpenAIConf{Adapter: "cli", Target: "codex"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"qwen3.6:35b",
"messages":[{"role":"user","content":"push changes"}],
@ -379,6 +404,79 @@ func TestChatCompletionsSynthesizesToolCallsFromClineTemplateCall(t *testing.T)
}
}
func TestChatCompletionsDoesNotSynthesizeTextToolCallsForProviderRoute(t *testing.T) {
fake := &fakeRunService{events: make(chan *iop.RunEvent, 2)}
fake.events <- &iop.RunEvent{Type: "delta", Delta: "먼저 현재 git 상태를 확인하겠습니다.\n\n<tool_call>\n<function=run_commands>\n<parameter=commands>\n[\"git status\"]\n</parameter>\n</function>\n</tool_call>"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"qwen3.6:35b",
"messages":[{"role":"user","content":"status"}],
"tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object"}}}],
"tool_choice":"auto"
}`))
w := httptest.NewRecorder()
srv.handleChatCompletions(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
}
var resp chatCompletionResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("decode response: %v", err)
}
choice := resp.Choices[0]
if choice.FinishReason == "tool_calls" || len(choice.Message.ToolCalls) != 0 {
t.Fatalf("provider text must not be synthesized: %+v", choice)
}
if !strings.Contains(choice.Message.Content, "<tool_call>") {
t.Fatalf("provider raw content should be preserved: %q", choice.Message.Content)
}
}
func TestChatCompletionsPassesThroughNativeToolCallsFromProviderRoute(t *testing.T) {
fake := &fakeRunService{events: make(chan *iop.RunEvent, 1)}
fake.events <- &iop.RunEvent{
Type: "complete",
Metadata: map[string]string{
"finish_reason": "tool_calls",
runtimeMetadataOpenAIToolCalls: `[{"id":"call_1","type":"function","function":{"name":"run_commands","arguments":"{\"commands\":[\"git status\"]}"}}]`,
},
}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"qwen3.6:35b",
"messages":[{"role":"user","content":"status"}],
"tools":[{"type":"function","function":{"name":"run_commands"}}],
"tool_choice":"auto"
}`))
w := httptest.NewRecorder()
srv.handleChatCompletions(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
}
var resp chatCompletionResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("decode response: %v", err)
}
choice := resp.Choices[0]
if choice.FinishReason != "tool_calls" {
t.Fatalf("finish_reason: got %q, want tool_calls", choice.FinishReason)
}
if len(choice.Message.ToolCalls) != 1 {
t.Fatalf("tool_calls len: got %d", len(choice.Message.ToolCalls))
}
call := choice.Message.ToolCalls[0].(map[string]any)
if call["id"] != "call_1" {
t.Fatalf("tool call id: got %+v", call["id"])
}
}
func TestSynthesizesTemplateToolCallConvertsCommandObjectsToShellStrings(t *testing.T) {
content := "확인합니다.\n\n{{run_commands(commands=[{'command': 'command -v git && echo \"found\" || echo \"not found\"', 'runInTerminal': True}])}}"
tools := []any{map[string]any{
@ -657,7 +755,7 @@ func TestChatCompletionsLeavesUnknownTextToolCallAsContent(t *testing.T) {
fake.events <- &iop.RunEvent{Type: "delta", Delta: "<tool_call><function=delete_everything><parameter=confirm>true</parameter></function></tool_call>"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
srv := NewServer(config.EdgeOpenAIConf{Adapter: "cli", Target: "codex"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"qwen3.6:35b",
"messages":[{"role":"user","content":"status"}],
@ -687,7 +785,7 @@ func TestChatCompletionsLeavesUnknownTemplateToolCallAsContent(t *testing.T) {
fake.events <- &iop.RunEvent{Type: "delta", Delta: "{{delete_everything(confirm=True)}}"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
srv := NewServer(config.EdgeOpenAIConf{Adapter: "cli", Target: "codex"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"qwen3.6:35b",
"messages":[{"role":"user","content":"status"}],
@ -772,7 +870,7 @@ func TestChatCompletionsStreamSynthesizesToolCallsFromClineTextBlock(t *testing.
fake.events <- &iop.RunEvent{Type: "delta", Delta: "<tool_call>\n<function=run_commands>\n<parameter=commands>\n[\"cd /config/workspace/iop && git status\"]\n</parameter>\n</function>\n</tool_call>"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
srv := NewServer(config.EdgeOpenAIConf{Adapter: "cli", Target: "codex"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"stream-model",
"stream":true,
@ -807,7 +905,7 @@ func TestChatCompletionsStreamSynthesizesToolCallsFromClineTemplateCall(t *testi
fake.events <- &iop.RunEvent{Type: "delta", Delta: "{{run_commands(commands=[{'command': 'git status', 'description': '현재 git 상태 확인', 'runInTerminal': True}]})}}"}
fake.events <- &iop.RunEvent{Type: "complete"}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
srv := NewServer(config.EdgeOpenAIConf{Adapter: "cli", Target: "codex"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"stream-model",
"stream":true,
@ -843,6 +941,41 @@ func TestChatCompletionsStreamSynthesizesToolCallsFromClineTemplateCall(t *testi
}
}
func TestChatCompletionsStreamPassesThroughNativeToolCallsFromProviderRoute(t *testing.T) {
fake := &fakeRunService{events: make(chan *iop.RunEvent, 2)}
fake.events <- &iop.RunEvent{Type: "delta", Delta: "checking"}
fake.events <- &iop.RunEvent{
Type: "complete",
Metadata: map[string]string{
"finish_reason": "tool_calls",
runtimeMetadataOpenAIToolCalls: `[{"id":"call_1","type":"function","function":{"name":"run_commands","arguments":"{\"commands\":[\"git status\"]}"}}]`,
},
}
srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
"model":"stream-model",
"stream":true,
"messages":[{"role":"user","content":"status"}],
"tools":[{"type":"function","function":{"name":"run_commands"}}],
"tool_choice":"auto"
}`))
w := httptest.NewRecorder()
srv.handleChatCompletions(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
}
body := w.Body.String()
if !strings.Contains(body, `"tool_calls"`) || !strings.Contains(body, `"call_1"`) {
t.Fatalf("stream did not pass through native tool_calls:\n%s", body)
}
if !strings.Contains(body, `"finish_reason":"tool_calls"`) {
t.Fatalf("stream finish_reason:\n%s", body)
}
}
func TestChatCompletionsStreamingReportsClosedRunStream(t *testing.T) {
fake := &fakeRunService{events: make(chan *iop.RunEvent)}
close(fake.events)
@ -1669,7 +1802,7 @@ func TestCollectRunResultTimesOut(t *testing.T) {
}
ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond)
defer cancel()
_, _, _, _, err := collectRunResult(ctx, handle.Stream(), handle.WaitTimeout())
_, _, _, _, _, err := collectRunResult(ctx, handle.Stream(), handle.WaitTimeout())
if err == nil {
t.Fatal("expected timeout error")
}
@ -1958,7 +2091,7 @@ func TestCollectRunResultFailsWhenEventStreamCloses(t *testing.T) {
},
}
_, _, _, _, err := collectRunResult(context.Background(), handle.Stream(), handle.WaitTimeout())
_, _, _, _, _, err := collectRunResult(context.Background(), handle.Stream(), handle.WaitTimeout())
if err == nil {
t.Fatal("expected closed stream error")
}

48
apps/edge/internal/openai/stream.go
View file

@ -153,7 +153,7 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re
}
func (s *Server) streamBufferedChatCompletion(w http.ResponseWriter, r *http.Request, handle edgeservice.RunResult, id string, created int64, model string, flusher http.Flusher, outputPolicy strictOutputPolicy, tools []any) {
text, reasoning, finishReason, _, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
text, reasoning, finishReason, nativeToolCalls, _, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
if err != nil {
writeSSEError(w, flusher, err.Error())
return
@ -170,8 +170,8 @@ func (s *Server) streamBufferedChatCompletion(w http.ResponseWriter, r *http.Req
zap.String("content_preview", previewString(text, 1000)),
zap.String("reasoning_preview", previewString(reasoning, 1000)),
)
if cleaned, toolCalls := synthesizeToolCallsFromText(text, tools, handle.Dispatch().RunID); len(toolCalls) > 0 {
if cleaned != "" {
if len(nativeToolCalls) > 0 {
if text != "" {
writeSSE(w, flusher, chatCompletionChunk{
ID: id,
Object: "chat.completion.chunk",
@ -179,7 +179,7 @@ func (s *Server) streamBufferedChatCompletion(w http.ResponseWriter, r *http.Req
Model: model,
Choices: []chatCompletionChunkChoice{{
Index: 0,
Delta: chatDelta{Content: cleaned},
Delta: chatDelta{Content: text},
}},
})
}
@ -190,10 +190,48 @@ func (s *Server) streamBufferedChatCompletion(w http.ResponseWriter, r *http.Req
Model: model,
Choices: []chatCompletionChunkChoice{{
Index: 0,
Delta: chatDelta{ToolCalls: toolCallsForStreamDelta(toolCalls)},
Delta: chatDelta{ToolCalls: toolCallsForStreamDelta(nativeToolCalls)},
}},
})
finishReason = "tool_calls"
} else if shouldSynthesizeTextToolCalls(handle.Dispatch()) {
cleaned, toolCalls := synthesizeToolCallsFromText(text, tools, handle.Dispatch().RunID)
if len(toolCalls) > 0 {
if cleaned != "" {
writeSSE(w, flusher, chatCompletionChunk{
ID: id,
Object: "chat.completion.chunk",
Created: created,
Model: model,
Choices: []chatCompletionChunkChoice{{
Index: 0,
Delta: chatDelta{Content: cleaned},
}},
})
}
writeSSE(w, flusher, chatCompletionChunk{
ID: id,
Object: "chat.completion.chunk",
Created: created,
Model: model,
Choices: []chatCompletionChunkChoice{{
Index: 0,
Delta: chatDelta{ToolCalls: toolCallsForStreamDelta(toolCalls)},
}},
})
finishReason = "tool_calls"
} else if text != "" {
writeSSE(w, flusher, chatCompletionChunk{
ID: id,
Object: "chat.completion.chunk",
Created: created,
Model: model,
Choices: []chatCompletionChunkChoice{{
Index: 0,
Delta: chatDelta{Content: text},
}},
})
}
} else if text != "" {
writeSSE(w, flusher, chatCompletionChunk{
ID: id,

18
apps/edge/internal/openai/types.go
View file

@ -62,7 +62,7 @@ func (m *chatMessage) UnmarshalJSON(b []byte) error {
return nil
}
func (req chatCompletionRequest) runInput(prompt string, messages []chatMessage, strictOutput bool) map[string]any {
func (req chatCompletionRequest) runInput(prompt string, messages []chatMessage, strictOutput, nativeToolCalls bool) map[string]any {
input := map[string]any{
"prompt": prompt,
"messages": chatMessagesInput(messages),
@ -92,19 +92,17 @@ func (req chatCompletionRequest) runInput(prompt string, messages []chatMessage,
}
if len(req.Tools) > 0 {
input["tools"] = req.Tools
input["tool_choice"] = normalizeToolChoice(req.ToolChoice)
if nativeToolCalls {
if req.ToolChoice != nil {
input["tool_choice"] = req.ToolChoice
}
} else {
input["tool_choice"] = "none"
}
}
return input
}
func normalizeToolChoice(any) string {
// IOP can synthesize OpenAI tool_calls from recognized text tool-call
// blocks, but it still cannot carry backend-native tool calls through the
// runtime event stream. Keep downstream auto tool selection disabled so a
// provider does not return structured calls that would be dropped.
return "none"
}
func setOptionInt(options map[string]any, key string, val *int) {
if val != nil {
options[key] = *val

16
apps/node/internal/adapters/ollama/ollama.go
View file

@ -20,6 +20,8 @@ import (
const Name = "ollama"
const runtimeMetadataOpenAIToolCalls = "openai_tool_calls"
type Ollama struct {
instanceName string
baseURL string
@ -159,6 +161,7 @@ func (o *Ollama) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink r
decoder := json.NewDecoder(resp.Body)
outputTokens := 0
var toolCalls []any
for {
var chunk ollamaChatChunk
if err := decoder.Decode(&chunk); err != nil {
@ -193,6 +196,9 @@ func (o *Ollama) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink r
return err
}
}
if len(chunk.Message.ToolCalls) > 0 {
toolCalls = append(toolCalls, chunk.Message.ToolCalls...)
}
if chunk.Done {
inputTokens := chunk.PromptEvalCount
if inputTokens == 0 {
@ -201,6 +207,15 @@ func (o *Ollama) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink r
if chunk.EvalCount > 0 {
outputTokens = chunk.EvalCount
}
var metadata map[string]string
if len(toolCalls) > 0 {
if encoded, err := json.Marshal(toolCalls); err == nil {
metadata = map[string]string{
"finish_reason": "tool_calls",
runtimeMetadataOpenAIToolCalls: string(encoded),
}
}
}
return sink.Emit(ctx, runtime.RuntimeEvent{
RunID: spec.RunID,
Type: runtime.EventTypeComplete,
@ -209,6 +224,7 @@ func (o *Ollama) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink r
InputTokens: inputTokens,
OutputTokens: outputTokens,
},
Metadata: metadata,
Timestamp: time.Now(),
})
}

35
apps/node/internal/adapters/ollama/ollama_test.go
View file

@ -163,6 +163,41 @@ func TestOllamaExecutePassesOptionsAndTopLevelFields(t *testing.T) {
}
}
func TestOllamaExecutePreservesNativeToolCalls(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/chat" {
t.Fatalf("unexpected path %s", r.URL.Path)
}
w.Header().Set("Content-Type", "application/x-ndjson")
_, _ = w.Write([]byte(`{"message":{"role":"assistant","tool_calls":[{"id":"call_1","type":"function","function":{"name":"run_commands","arguments":"{\"commands\":[\"git status\"]}"}}]},"done":false}` + "\n"))
_, _ = w.Write([]byte(`{"done":true}` + "\n"))
}))
defer server.Close()
adapter := New(config.OllamaConf{BaseURL: server.URL}, zap.NewNop())
sink := &fakeSink{}
err := adapter.Execute(context.Background(), noderuntime.ExecutionSpec{
RunID: "run-tools",
Target: "llama-test",
Input: map[string]any{"prompt": "status"},
}, sink)
if err != nil {
t.Fatalf("Execute failed: %v", err)
}
events := sink.all()
complete := events[len(events)-1]
if complete.Metadata["finish_reason"] != "tool_calls" {
t.Fatalf("finish_reason: %+v", complete.Metadata)
}
var toolCalls []map[string]any
if err := json.Unmarshal([]byte(complete.Metadata[runtimeMetadataOpenAIToolCalls]), &toolCalls); err != nil {
t.Fatalf("tool_calls metadata JSON: %v", err)
}
if len(toolCalls) != 1 || toolCalls[0]["id"] != "call_1" {
t.Fatalf("tool_calls: %+v", toolCalls)
}
}
func TestOllamaExecuteEmitsErrorForHTTPFailure(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
http.Error(w, "boom", http.StatusBadGateway)

136
apps/node/internal/adapters/openai_compat/openai_compat.go
View file

@ -25,6 +25,8 @@ import (
const Name = "openai_compat"
const runtimeMetadataOpenAIToolCalls = "openai_tool_calls"
type Adapter struct {
instanceName string
provider string
@ -156,6 +158,7 @@ func (a *Adapter) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink
outputTokens := 0
finishReason := ""
var usage *runtime.UsageStats
var toolCalls openAIToolCallAccumulator
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
@ -163,7 +166,7 @@ func (a *Adapter) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink
}
payload := strings.TrimPrefix(line, "data: ")
if payload == "[DONE]" {
return sink.Emit(ctx, completeEvent(spec.RunID, finishReason, usage, outputTokens))
return sink.Emit(ctx, completeEvent(spec.RunID, finishReason, usage, outputTokens, toolCalls.ToolCalls()))
}
var chunk chatChunk
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
@ -182,6 +185,9 @@ func (a *Adapter) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink
if choice.FinishReason != nil && *choice.FinishReason != "" {
finishReason = *choice.FinishReason
}
if len(choice.Delta.ToolCalls) > 0 {
toolCalls.AddDelta(choice.Delta.ToolCalls)
}
if reasoning := choice.Delta.ReasoningContent; reasoning != "" {
if err := sink.Emit(ctx, runtime.RuntimeEvent{
RunID: spec.RunID,
@ -327,7 +333,7 @@ func buildRequestBody(model string, messages []chatMessage, input map[string]any
return body
}
func completeEvent(runID, finishReason string, usage *runtime.UsageStats, outputTokens int) runtime.RuntimeEvent {
func completeEvent(runID, finishReason string, usage *runtime.UsageStats, outputTokens int, toolCalls []any) runtime.RuntimeEvent {
if usage == nil {
usage = &runtime.UsageStats{OutputTokens: outputTokens}
}
@ -335,6 +341,17 @@ func completeEvent(runID, finishReason string, usage *runtime.UsageStats, output
if finishReason != "" {
metadata = map[string]string{"finish_reason": finishReason}
}
if len(toolCalls) > 0 {
if metadata == nil {
metadata = make(map[string]string, 2)
}
if finishReason == "" {
metadata["finish_reason"] = "tool_calls"
}
if encoded, err := json.Marshal(toolCalls); err == nil {
metadata[runtimeMetadataOpenAIToolCalls] = string(encoded)
}
}
return runtime.RuntimeEvent{
RunID: runID,
Type: runtime.EventTypeComplete,
@ -439,6 +456,7 @@ type chatChunk struct {
Delta struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
ToolCalls []any `json:"tool_calls"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
@ -448,6 +466,120 @@ type chatChunk struct {
} `json:"usage"`
}
type openAIToolCallAccumulator struct {
calls map[int]map[string]any
order []int
}
func (a *openAIToolCallAccumulator) AddDelta(raw []any) {
for fallbackIndex, item := range raw {
call, ok := item.(map[string]any)
if !ok {
continue
}
index := fallbackIndex
if parsed, ok := numericIndex(call["index"]); ok {
index = parsed
}
dst := a.ensure(index)
for key, value := range call {
switch key {
case "index":
continue
case "function":
mergeToolCallFunction(dst, value)
default:
if !emptyToolCallValue(value) {
dst[key] = value
}
}
}
}
}
func (a *openAIToolCallAccumulator) ToolCalls() []any {
if len(a.order) == 0 {
return nil
}
out := make([]any, 0, len(a.order))
for _, index := range a.order {
call := a.calls[index]
if len(call) == 0 {
continue
}
copyCall := make(map[string]any, len(call))
for key, value := range call {
copyCall[key] = value
}
out = append(out, copyCall)
}
return out
}
func (a *openAIToolCallAccumulator) ensure(index int) map[string]any {
if a.calls == nil {
a.calls = make(map[int]map[string]any)
}
if call, ok := a.calls[index]; ok {
return call
}
call := make(map[string]any)
a.calls[index] = call
a.order = append(a.order, index)
return call
}
func mergeToolCallFunction(call map[string]any, value any) {
fn, ok := value.(map[string]any)
if !ok {
return
}
current, _ := call["function"].(map[string]any)
if current == nil {
current = make(map[string]any)
call["function"] = current
}
for key, item := range fn {
if key == "arguments" {
if part, ok := item.(string); ok {
current["arguments"] = currentString(current["arguments"]) + part
continue
}
}
if !emptyToolCallValue(item) {
current[key] = item
}
}
}
func numericIndex(value any) (int, bool) {
switch v := value.(type) {
case int:
return v, true
case float64:
return int(v), true
default:
return 0, false
}
}
func emptyToolCallValue(value any) bool {
if value == nil {
return true
}
if text, ok := value.(string); ok {
return text == ""
}
return false
}
func currentString(value any) string {
if text, ok := value.(string); ok {
return text
}
return ""
}
type modelsResponse struct {
Data []struct {
ID string `json:"id"`

36
apps/node/internal/adapters/openai_compat/openai_compat_test.go
View file

@ -376,6 +376,42 @@ func TestOpenAICompatExecutePreservesToolCallMessages(t *testing.T) {
}
}
func TestOpenAICompatExecutePreservesNativeToolCalls(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
_, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"run_commands","arguments":"{\"commands\":"}}]}}]}`)
_, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"[\"git status\"]}"}}]},"finish_reason":"tool_calls"}]}`)
_, _ = fmt.Fprintf(w, "data: [DONE]\n\n")
}))
defer server.Close()
adapter := New(config.OpenAICompatConf{Endpoint: server.URL}, zap.NewNop())
sink := &fakeSink{}
if err := adapter.Execute(context.Background(), runtime.ExecutionSpec{
RunID: "run-native-tools",
Target: "qwen3.6:35b",
Input: map[string]any{"prompt": "status"},
}, sink); err != nil {
t.Fatalf("Execute failed: %v", err)
}
events := sink.all()
complete := events[len(events)-1]
if complete.Metadata["finish_reason"] != "tool_calls" {
t.Fatalf("finish_reason: %+v", complete.Metadata)
}
var toolCalls []map[string]any
if err := json.Unmarshal([]byte(complete.Metadata[runtimeMetadataOpenAIToolCalls]), &toolCalls); err != nil {
t.Fatalf("tool_calls metadata JSON: %v", err)
}
if len(toolCalls) != 1 || toolCalls[0]["id"] != "call_1" {
t.Fatalf("tool_calls: %+v", toolCalls)
}
fn := toolCalls[0]["function"].(map[string]any)
if fn["name"] != "run_commands" || fn["arguments"] != `{"commands":["git status"]}` {
t.Fatalf("function: %+v", fn)
}
}
func TestOpenAICompatExecuteRejectsEmptyEndpointOrModel(t *testing.T) {
t.Run("empty_endpoint", func(t *testing.T) {
adapter := New(config.OpenAICompatConf{}, zap.NewNop())

220
apps/node/internal/adapters/vllm/vllm.go
View file

@ -22,6 +22,8 @@ import (
const Name = "vllm"
const runtimeMetadataOpenAIToolCalls = "openai_tool_calls"
type Vllm struct {
instanceName string
endpoint string
@ -111,6 +113,12 @@ func (v *Vllm) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink run
Messages: messages,
Stream: true,
}
if v, ok := spec.Input["tools"]; ok {
chatReq.Tools = v
}
if v, ok := spec.Input["tool_choice"]; ok {
chatReq.ToolChoice = v
}
body, err := json.Marshal(chatReq)
if err != nil {
return fmt.Errorf("vllm adapter: marshal request: %w", err)
@ -143,6 +151,9 @@ func (v *Vllm) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink run
scanner := bufio.NewScanner(resp.Body)
outputTokens := 0
finishReason := ""
var usage *runtime.UsageStats
var toolCalls openAIToolCallAccumulator
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
@ -150,15 +161,7 @@ func (v *Vllm) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink run
}
payload := strings.TrimPrefix(line, "data: ")
if payload == "[DONE]" {
return sink.Emit(ctx, runtime.RuntimeEvent{
RunID: spec.RunID,
Type: runtime.EventTypeComplete,
Message: "vllm chat complete",
Usage: &runtime.UsageStats{
OutputTokens: outputTokens,
},
Timestamp: time.Now(),
})
return sink.Emit(ctx, completeEvent(spec.RunID, finishReason, usage, outputTokens, toolCalls.ToolCalls()))
}
var chunk vllmChatChunk
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
@ -167,7 +170,20 @@ func (v *Vllm) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink run
if len(chunk.Choices) == 0 {
continue
}
content := chunk.Choices[0].Delta.Content
if chunk.Usage != nil {
usage = &runtime.UsageStats{
InputTokens: chunk.Usage.PromptTokens,
OutputTokens: chunk.Usage.CompletionTokens,
}
}
choice := chunk.Choices[0]
if choice.FinishReason != nil && *choice.FinishReason != "" {
finishReason = *choice.FinishReason
}
if len(choice.Delta.ToolCalls) > 0 {
toolCalls.AddDelta(choice.Delta.ToolCalls)
}
content := choice.Delta.Content
if content != "" {
outputTokens += len(strings.Fields(content))
if err := sink.Emit(ctx, runtime.RuntimeEvent{
@ -272,12 +288,20 @@ func messagesFromInput(input map[string]any) []vllmMessage {
}
role, _ := m["role"].(string)
content, _ := m["content"].(string)
toolCallID, _ := m["tool_call_id"].(string)
toolCalls := anySlice(m["tool_calls"])
role = strings.TrimSpace(role)
content = strings.TrimSpace(content)
if role == "" || content == "" {
toolCallID = strings.TrimSpace(toolCallID)
if role == "" || (content == "" && len(toolCalls) == 0) {
continue
}
out = append(out, vllmMessage{Role: role, Content: content})
out = append(out, vllmMessage{
Role: role,
Content: content,
ToolCalls: toolCalls,
ToolCallID: toolCallID,
})
}
if len(out) > 0 {
return out
@ -298,6 +322,42 @@ func emitError(ctx context.Context, sink runtime.EventSink, runID, msg string) e
})
}
func completeEvent(runID, finishReason string, usage *runtime.UsageStats, outputTokens int, toolCalls []any) runtime.RuntimeEvent {
if usage == nil {
usage = &runtime.UsageStats{OutputTokens: outputTokens}
}
var metadata map[string]string
if finishReason != "" {
metadata = map[string]string{"finish_reason": finishReason}
}
if len(toolCalls) > 0 {
if metadata == nil {
metadata = make(map[string]string, 2)
}
if finishReason == "" {
metadata["finish_reason"] = "tool_calls"
}
if encoded, err := json.Marshal(toolCalls); err == nil {
metadata[runtimeMetadataOpenAIToolCalls] = string(encoded)
}
}
return runtime.RuntimeEvent{
RunID: runID,
Type: runtime.EventTypeComplete,
Message: "vllm chat complete",
Usage: usage,
Metadata: metadata,
Timestamp: time.Now(),
}
}
func anySlice(v any) []any {
if items, ok := v.([]any); ok {
return items
}
return nil
}
func stringInput(input map[string]any, key string) string {
if input == nil {
return ""
@ -323,22 +383,146 @@ func readLimited(r io.Reader, limit int64) string {
}
type vllmChatRequest struct {
Model string `json:"model"`
Messages []vllmMessage `json:"messages"`
Stream bool `json:"stream"`
Model string `json:"model"`
Messages []vllmMessage `json:"messages"`
Stream bool `json:"stream"`
Tools any `json:"tools,omitempty"`
ToolChoice any `json:"tool_choice,omitempty"`
}
type vllmMessage struct {
Role string `json:"role"`
Content string `json:"content"`
Role string `json:"role"`
Content string `json:"content"`
ToolCalls []any `json:"tool_calls,omitempty"`
ToolCallID string `json:"tool_call_id,omitempty"`
}
type vllmChatChunk struct {
Choices []struct {
Delta struct {
Content string `json:"content"`
Content string `json:"content"`
ToolCalls []any `json:"tool_calls"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
Usage *struct {
PromptTokens int `json:"prompt_tokens"`
CompletionTokens int `json:"completion_tokens"`
} `json:"usage"`
}
type openAIToolCallAccumulator struct {
calls map[int]map[string]any
order []int
}
func (a *openAIToolCallAccumulator) AddDelta(raw []any) {
for fallbackIndex, item := range raw {
call, ok := item.(map[string]any)
if !ok {
continue
}
index := fallbackIndex
if parsed, ok := numericIndex(call["index"]); ok {
index = parsed
}
dst := a.ensure(index)
for key, value := range call {
switch key {
case "index":
continue
case "function":
mergeToolCallFunction(dst, value)
default:
if !emptyToolCallValue(value) {
dst[key] = value
}
}
}
}
}
func (a *openAIToolCallAccumulator) ToolCalls() []any {
if len(a.order) == 0 {
return nil
}
out := make([]any, 0, len(a.order))
for _, index := range a.order {
call := a.calls[index]
if len(call) == 0 {
continue
}
copyCall := make(map[string]any, len(call))
for key, value := range call {
copyCall[key] = value
}
out = append(out, copyCall)
}
return out
}
func (a *openAIToolCallAccumulator) ensure(index int) map[string]any {
if a.calls == nil {
a.calls = make(map[int]map[string]any)
}
if call, ok := a.calls[index]; ok {
return call
}
call := make(map[string]any)
a.calls[index] = call
a.order = append(a.order, index)
return call
}
func mergeToolCallFunction(call map[string]any, value any) {
fn, ok := value.(map[string]any)
if !ok {
return
}
current, _ := call["function"].(map[string]any)
if current == nil {
current = make(map[string]any)
call["function"] = current
}
for key, item := range fn {
if key == "arguments" {
if part, ok := item.(string); ok {
current["arguments"] = currentString(current["arguments"]) + part
continue
}
}
if !emptyToolCallValue(item) {
current[key] = item
}
}
}
func numericIndex(value any) (int, bool) {
switch v := value.(type) {
case int:
return v, true
case float64:
return int(v), true
default:
return 0, false
}
}
func emptyToolCallValue(value any) bool {
if value == nil {
return true
}
if text, ok := value.(string); ok {
return text == ""
}
return false
}
func currentString(value any) string {
if text, ok := value.(string); ok {
return text
}
return ""
}
type vllmModelsResponse struct {

49
apps/node/internal/adapters/vllm/vllm_test.go
View file

@ -142,6 +142,55 @@ func TestVllmExecuteStreamsDeltas(t *testing.T) {
}
}
func TestVllmExecutePassesToolsAndPreservesNativeToolCalls(t *testing.T) {
var gotToolChoice any
var gotTools []any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var req map[string]any
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
gotToolChoice = req["tool_choice"]
gotTools, _ = req["tools"].([]any)
w.Header().Set("Content-Type", "text/event-stream")
_, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"run_commands","arguments":"{\"commands\":"}}]}}]}`)
_, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"[\"git status\"]}"}}]},"finish_reason":"tool_calls"}]}`)
_, _ = fmt.Fprintf(w, "data: [DONE]\n\n")
}))
defer server.Close()
adapter := New(config.VllmConf{Endpoint: server.URL}, zap.NewNop())
sink := &fakeSink{}
err := adapter.Execute(context.Background(), runtime.ExecutionSpec{
RunID: "run-tools",
Target: "llama-3",
Input: map[string]any{
"prompt": "status",
"tools": []any{map[string]any{"type": "function"}},
"tool_choice": "auto",
},
}, sink)
if err != nil {
t.Fatalf("Execute failed: %v", err)
}
if gotToolChoice != "auto" || len(gotTools) != 1 {
t.Fatalf("tools/tool_choice not passed: tools=%+v choice=%+v", gotTools, gotToolChoice)
}
events := sink.all()
complete := events[len(events)-1]
if complete.Metadata["finish_reason"] != "tool_calls" {
t.Fatalf("finish_reason: %+v", complete.Metadata)
}
var toolCalls []map[string]any
if err := json.Unmarshal([]byte(complete.Metadata[runtimeMetadataOpenAIToolCalls]), &toolCalls); err != nil {
t.Fatalf("tool_calls metadata JSON: %v", err)
}
fn := toolCalls[0]["function"].(map[string]any)
if fn["arguments"] != `{"commands":["git status"]}` {
t.Fatalf("arguments: %+v", fn["arguments"])
}
}
func TestVllmExecuteUsesMessagesInput(t *testing.T) {
var gotMessages []map[string]any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {