fix(llm): strip inline <think> reasoning from Dahl/MiniMax streams; dedupe identical parallel tool calls; hermetic config tests
This commit is contained in:
parent
46d98502dd
commit
546a9315d5
6 changed files with 264 additions and 5 deletions
|
|
@ -178,11 +178,20 @@ func (a *Agent) Run(parent context.Context, sessionID, userText string, emit fun
|
|||
if len(resp.ToolCalls) == 0 {
|
||||
return resp.Content, nil
|
||||
}
|
||||
// Some models (seen with GLM/DeepSeek on Dahl) emit the same call several times in
|
||||
// parallel; identical name+arguments run once and share the result.
|
||||
seen := map[string]string{}
|
||||
for _, tc := range resp.ToolCalls {
|
||||
emit(Event{Type: "tool_call", Tool: tc.Name, Args: tc.Arguments, CallID: tc.ID})
|
||||
out := "error: cancelled by user"
|
||||
if !cancelled() {
|
||||
key := tc.Name + "\x00" + tc.Arguments
|
||||
out, dup := seen[key]
|
||||
switch {
|
||||
case dup:
|
||||
case cancelled():
|
||||
out = "error: cancelled by user"
|
||||
default:
|
||||
out = a.Tools.Call(ctx, tc.Name, tc.Arguments)
|
||||
seen[key] = out
|
||||
}
|
||||
emit(Event{Type: "tool_result", Tool: tc.Name, Text: clip(out, 2000), CallID: tc.ID})
|
||||
// Every tool call gets a result so the history stays valid for the provider.
|
||||
|
|
|
|||
|
|
@ -206,3 +206,29 @@ func TestSummarizeNowManual(t *testing.T) {
|
|||
t.Fatalf("%v %v", sm, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDuplicateToolCallsRunOnce(t *testing.T) {
|
||||
st := newStore(t)
|
||||
ss, _ := st.CreateSession(context.Background(), "")
|
||||
n := 0
|
||||
now := func() time.Time { n++; return time.Date(2026, 10, 25, 6, 0, 0, 0, time.UTC) }
|
||||
dup := []llm.ToolCall{{ID: "a", Name: "time_now", Arguments: `{"timezone":"UTC"}`}, {ID: "b", Name: "time_now", Arguments: `{"timezone":"UTC"}`}, {ID: "c", Name: "time_now", Arguments: `{}`}}
|
||||
p := &scripted{resps: []llm.Response{{ToolCalls: dup}, {Content: "ok"}}}
|
||||
a := &Agent{LLM: p, Store: st, Tools: tools.NewRegistry(tools.TimeNow{Now: now}), MaxIterations: 3}
|
||||
if _, err := a.Run(context.Background(), ss.ID, "zeit?", nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 2 {
|
||||
t.Fatalf("tool executed %d times, want 2", n)
|
||||
}
|
||||
msgs, _ := st.Messages(context.Background(), ss.ID, 0)
|
||||
tools := 0
|
||||
for _, m := range msgs {
|
||||
if m.Role == llm.RoleTool {
|
||||
tools++
|
||||
}
|
||||
}
|
||||
if tools != 3 {
|
||||
t.Fatalf("every call id needs a result, got %d", tools)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import (
|
|||
)
|
||||
|
||||
func load(t *testing.T, yaml string) (Config, error) {
|
||||
t.Helper()
|
||||
t.Helper()
|
||||
p := filepath.Join(t.TempDir(), "c.yaml")
|
||||
os.WriteFile(p, []byte("auth_token: aaaaaaaaaaaaaaaaaaaa\n"+yaml), 0o600)
|
||||
|
|
@ -74,3 +75,13 @@ func TestAllPresetsComplete(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestMain keeps tests hermetic when the developer shell has BYTECLAW_* or provider keys set.
|
||||
func TestMain(m *testing.M) {
|
||||
for _, k := range []string{"BYTECLAW_LISTEN", "BYTECLAW_DATA_DIR", "BYTECLAW_WORKSPACE", "BYTECLAW_LLM_PROVIDER",
|
||||
"BYTECLAW_LLM_BASE_URL", "BYTECLAW_LLM_API_KEY", "BYTECLAW_LLM_MODEL", "BYTECLAW_ALLOWED_ORIGINS", "BYTECLAW_AUTH_TOKEN",
|
||||
"OPENAI_API_KEY", "OPENROUTER_API_KEY", "DAHL_API_KEY", "GONKAROUTER_API_KEY"} {
|
||||
os.Unsetenv(k)
|
||||
}
|
||||
os.Exit(m.Run())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -131,6 +131,7 @@ type partialCall struct {
|
|||
func parseStream(r io.Reader, onDelta DeltaFunc) (Response, error) {
|
||||
var out Response
|
||||
var text strings.Builder
|
||||
var tf thinkFilter
|
||||
calls := map[int]*partialCall{}
|
||||
sc := bufio.NewScanner(r)
|
||||
sc.Buffer(make([]byte, 64*1024), 4<<20)
|
||||
|
|
@ -152,9 +153,11 @@ func parseStream(r io.Reader, onDelta DeltaFunc) (Response, error) {
|
|||
}
|
||||
for _, c := range ch.Choices {
|
||||
if c.Delta.Content != "" {
|
||||
text.WriteString(c.Delta.Content)
|
||||
if v := tf.push(c.Delta.Content); v != "" {
|
||||
text.WriteString(v)
|
||||
if onDelta != nil {
|
||||
onDelta(c.Delta.Content)
|
||||
onDelta(v)
|
||||
}
|
||||
}
|
||||
}
|
||||
for i, tc := range c.Delta.ToolCalls {
|
||||
|
|
@ -183,6 +186,12 @@ func parseStream(r io.Reader, onDelta DeltaFunc) (Response, error) {
|
|||
if err := sc.Err(); err != nil {
|
||||
return out, fmt.Errorf("read stream: %w", err)
|
||||
}
|
||||
if v := tf.finish(); v != "" {
|
||||
text.WriteString(v)
|
||||
if onDelta != nil {
|
||||
onDelta(v)
|
||||
}
|
||||
}
|
||||
out.Content = text.String()
|
||||
idxs := make([]int, 0, len(calls))
|
||||
for i := range calls {
|
||||
|
|
|
|||
129
internal/llm/think.go
Normal file
129
internal/llm/think.go
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
package llm
|
||||
|
||||
import "strings"
|
||||
|
||||
// thinkFilter removes inline reasoning blocks ("<think>…</think>") that some
|
||||
// OpenAI-compatible backends (e.g. MiniMax-M2.x on vLLM, seen on Dahl) put into
|
||||
// the normal content stream instead of a separate reasoning field. Tags may be
|
||||
// split across chunks. Text inside a block is never streamed to the user.
|
||||
type thinkFilter struct {
|
||||
pending string // undecided tail that may be the start of a tag
|
||||
inThink bool
|
||||
think strings.Builder // reasoning text of the current (unclosed) block
|
||||
out strings.Builder // visible content
|
||||
sawOpen bool
|
||||
}
|
||||
|
||||
const (
|
||||
openTag = "<think>"
|
||||
closeTag = "</think>"
|
||||
gap = "\n\n\n"
|
||||
)
|
||||
|
||||
// push consumes a chunk and returns the text that may be shown now.
|
||||
func (f *thinkFilter) push(s string) string {
|
||||
buf := f.pending + s
|
||||
f.pending = ""
|
||||
var vis strings.Builder
|
||||
for buf != "" {
|
||||
tag := openTag
|
||||
if f.inThink {
|
||||
tag = closeTag
|
||||
}
|
||||
i, n := strings.Index(buf, tag), len(tag)
|
||||
// Dahl/vLLM sometimes swallows "</think>" and leaves only a "\n\n\n" gap;
|
||||
// treat that gap as the end of the block so the answer still streams live.
|
||||
if f.inThink {
|
||||
if j := strings.Index(buf, gap); j >= 0 && (i < 0 || j < i) {
|
||||
i, n = j, len(gap)
|
||||
}
|
||||
}
|
||||
if i >= 0 {
|
||||
f.emit(&vis, buf[:i])
|
||||
buf = buf[i+n:]
|
||||
if f.inThink {
|
||||
f.think.Reset()
|
||||
} else {
|
||||
f.sawOpen = true
|
||||
}
|
||||
f.inThink = !f.inThink
|
||||
continue
|
||||
}
|
||||
// a stray closing tag outside a block is dropped
|
||||
if !f.inThink {
|
||||
if i := strings.Index(buf, closeTag); i >= 0 {
|
||||
f.emit(&vis, buf[:i])
|
||||
buf = buf[i+len(closeTag):]
|
||||
continue
|
||||
}
|
||||
}
|
||||
keep := partialSuffix(buf, tag)
|
||||
if f.inThink {
|
||||
if k := partialSuffix(buf, gap); k > keep {
|
||||
keep = k
|
||||
}
|
||||
} else {
|
||||
if k := partialSuffix(buf, closeTag); k > keep {
|
||||
keep = k
|
||||
}
|
||||
}
|
||||
f.emit(&vis, buf[:len(buf)-keep])
|
||||
f.pending = buf[len(buf)-keep:]
|
||||
buf = ""
|
||||
}
|
||||
return vis.String()
|
||||
}
|
||||
|
||||
func (f *thinkFilter) emit(vis *strings.Builder, s string) {
|
||||
if s == "" {
|
||||
return
|
||||
}
|
||||
if f.inThink {
|
||||
f.think.WriteString(s)
|
||||
return
|
||||
}
|
||||
// drop whitespace directly after a closed block at the start of the answer
|
||||
if f.out.Len() == 0 {
|
||||
s = strings.TrimLeft(s, " \n\r\t")
|
||||
if s == "" {
|
||||
return
|
||||
}
|
||||
}
|
||||
f.out.WriteString(s)
|
||||
vis.WriteString(s)
|
||||
}
|
||||
|
||||
// finish flushes the remainder. If a block was never closed (observed when the
|
||||
// backend swallows "</think>"), the answer is recovered from the text after the
|
||||
// first blank-line gap so the user does not end up with nothing.
|
||||
func (f *thinkFilter) finish() string {
|
||||
var vis strings.Builder
|
||||
if f.pending != "" {
|
||||
p := f.pending
|
||||
f.pending = ""
|
||||
f.emit(&vis, p)
|
||||
}
|
||||
if f.inThink {
|
||||
t := f.think.String()
|
||||
f.inThink = false
|
||||
f.think.Reset()
|
||||
ans := ""
|
||||
if i := strings.Index(t, "\n\n\n"); i >= 0 {
|
||||
ans = t[i:]
|
||||
} else if f.out.Len() == 0 {
|
||||
ans = t // better a visible reasoning than an empty reply
|
||||
}
|
||||
f.emit(&vis, strings.TrimSpace(ans))
|
||||
}
|
||||
return vis.String()
|
||||
}
|
||||
|
||||
// partialSuffix returns the length of the longest suffix of s that is a proper prefix of tag.
|
||||
func partialSuffix(s, tag string) int {
|
||||
for n := len(tag) - 1; n > 0; n-- {
|
||||
if len(s) >= n && strings.HasSuffix(s, tag[:n]) {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
75
internal/llm/think_test.go
Normal file
75
internal/llm/think_test.go
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
package llm
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func run(chunks ...string) (string, string) {
|
||||
var f thinkFilter
|
||||
var streamed strings.Builder
|
||||
for _, c := range chunks {
|
||||
streamed.WriteString(f.push(c))
|
||||
}
|
||||
streamed.WriteString(f.finish())
|
||||
return streamed.String(), f.out.String()
|
||||
}
|
||||
|
||||
func TestThinkFilter(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
chunks []string
|
||||
want string
|
||||
}{
|
||||
{"plain", []string{"Hallo ", "Welt"}, "Hallo Welt"},
|
||||
{"block", []string{"<think>reasoning\n</think>\n\nHallo!"}, "Hallo!"},
|
||||
{"split tags", []string{"<th", "ink>abc</th", "ink>", "\n\nHal", "lo"}, "Hallo"},
|
||||
{"char by char", strings.Split("<think>x</think>Ja", ""), "Ja"},
|
||||
{"stray close", []string{"Hallo!</think>\n\nHallo!"}, "Hallo!\n\nHallo!"},
|
||||
{"unclosed with gap (Dahl)", []string{"<think>The user asks.\n\n\nHallo! Wie geht's?"}, "Hallo! Wie geht's?"},
|
||||
{"gap streams live", []string{"<think>a\n", "\n", "\nAntw", "ort"}, "Antwort"},
|
||||
{"gap outside block kept", []string{"A\n\n\nB"}, "A\n\n\nB"},
|
||||
{"unclosed no gap", []string{"<think>nur Gedanken"}, "nur Gedanken"},
|
||||
{"lt not a tag", []string{"a <b> c <", "3"}, "a <b> c <3"},
|
||||
{"trailing partial", []string{"x <thi"}, "x <thi"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
streamed, out := run(c.chunks...)
|
||||
if streamed != c.want || out != c.want {
|
||||
t.Errorf("%s: streamed=%q out=%q want %q", c.name, streamed, out, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseStreamStripsThink(t *testing.T) {
|
||||
sse := `data: {"choices":[{"delta":{"content":"","role":"assistant"}}]}
|
||||
|
||||
data: {"choices":[{"delta":{"reasoning":"hidden"}}]}
|
||||
|
||||
data: {"choices":[{"delta":{"content":"<think>The user"}}]}
|
||||
|
||||
data: {"choices":[{"delta":{"content":" greets.\n</thi"}}]}
|
||||
|
||||
data: {"choices":[{"delta":{"content":"nk>\n\nHallo!"}}]}
|
||||
|
||||
data: {"choices":[{"delta":{},"finish_reason":"stop"}]}
|
||||
|
||||
data: [DONE]
|
||||
`
|
||||
var deltas []string
|
||||
resp, err := parseStream(strings.NewReader(sse), func(s string) { deltas = append(deltas, s) })
|
||||
if err != nil || resp.Content != "Hallo!" || strings.Join(deltas, "") != "Hallo!" {
|
||||
t.Fatalf("%q %v %v", resp.Content, deltas, err)
|
||||
}
|
||||
}
|
||||
|
||||
// The answer after a swallowed </think> must be streamed incrementally, not only at the end.
|
||||
func TestThinkGapStreamsIncrementally(t *testing.T) {
|
||||
var f thinkFilter
|
||||
if v := f.push("<think>Überlegung\n\n\nHal"); v != "Hal" {
|
||||
t.Fatalf("got %q", v)
|
||||
}
|
||||
if v := f.push("lo"); v != "lo" {
|
||||
t.Fatalf("got %q", v)
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue