Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions dashboard/frontend/src/types/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,7 @@ export interface DecisionModelSignal {
}
predicate?: NumericPredicate
timeout_ms?: number
prior_user_turns?: number
}

export interface ComplexityCandidates {
Expand Down
1 change: 1 addition & 0 deletions e2e/profiles/model-runtime/profile.go
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ func (p *Profile) GetTestCases() []string {
"model-runtime-embeddings-rerank",
"decision-runtime-routing",
"decision-runtime-set-span",
"decision-prior-user-turns",
"model-runtime-bundles",
"model-runtime-long-history",
"model-runtime-fail-open",
Expand Down
23 changes: 23 additions & 0 deletions e2e/profiles/model-runtime/values.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,25 @@ config:
- key: chat
description: Anything else
timeout_ms: 10000
# testcases/decision_prior_user_turns.go: reads the previous user turn
# too, so a conversation's follow-up is a separate call.
- name: follow_up_kind
description: Which kind of work the latest turn continues.
deployment: decision-fixture
prior_user_turns: 1
question:
type: choice
instructions: What kind of work does the latest user turn ask for?
choices:
- key: code
description: Writing, reviewing or debugging code
- key: math
description: Mathematics or quantitative reasoning
- key: writing
description: Prose, poems or other writing
- key: chat
description: Anything else
timeout_ms: 10000
- name: attached_answered
description: Matches whenever the attached decision model answers.
deployment: attached-decisions
Expand Down Expand Up @@ -254,6 +273,10 @@ config:
- {type: decision, name: request_kind, label: code}
- {type: decision, name: request_kind, label: math}
- {type: decision, name: request_kind, label: chat}
- {type: decision, name: follow_up_kind, label: code}
- {type: decision, name: follow_up_kind, label: math}
- {type: decision, name: follow_up_kind, label: writing}
- {type: decision, name: follow_up_kind, label: chat}
- {type: decision, name: topics, label: billing}
- {type: decision, name: topics, label: shipping}
- {type: decision, name: names, label: city}
Expand Down
107 changes: 107 additions & 0 deletions e2e/testcases/decision_prior_user_turns.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
package testcases

import (
"context"
"fmt"
"net/http"

"k8s.io/client-go/kubernetes"

"github.com/vllm-project/semantic-router/e2e/pkg/modelruntime"
pkgtestcases "github.com/vllm-project/semantic-router/e2e/pkg/testcases"
)

func init() {
pkgtestcases.Register("decision-prior-user-turns", pkgtestcases.TestCase{
Description: "A decision signal with prior_user_turns routes a follow-up on the runtime's answer for the earlier user turn and the follow-up together",
Tags: []string{"decision", "model-runtime", "routing", "history"},
Fn: testDecisionPriorUserTurns,
})
}

// The follow_up_kind question of e2e/profiles/model-runtime/values.yaml.
var decisionFollowUpQuestion = map[string]modelruntime.Question{
"follow_up_kind": {
Type: "choice",
Instructions: "What kind of work does the latest user turn ask for?",
Choices: []modelruntime.Choice{
{Key: "code", Description: "Writing, reviewing or debugging code"},
{Key: "math", Description: "Mathematics or quantitative reasoning"},
{Key: "writing", Description: "Prose, poems or other writing"},
{Key: "chat", Description: "Anything else"},
},
},
}

const decisionFollowUp = "Now make it shorter."

var decisionFirstTurns = []string{
"Write a Go function that merges two sorted slices.",
"Prove that the square root of two is irrational.",
"Write a short poem about rain on a tin roof.",
"Explain how a hash map handles collisions.",
"Draft an email asking my landlord to fix the heating.",
"Solve x squared minus five x plus six equals zero.",
}

// testDecisionPriorUserTurns finds a first turn for which the fixture runtime
// answers the follow-up differently alone and after that turn, so the route
// shows which state the Router asked. The fixture weights are random, so the
// runtime's own answers are the expectation.
func testDecisionPriorUserTurns(ctx context.Context, client *kubernetes.Clientset, opts pkgtestcases.TestCaseOptions) error {
session, err := openModelRuntimeSession(ctx, client, opts)
if err != nil {
return err
}
defer session.Close()
if err = session.waitReady(ctx, mrDecisionDeployment); err != nil {
return err
}
managed, _, err := session.managed(ctx, mrDecisionDeployment)
if err != nil {
return err
}
for _, first := range decisionFirstTurns {
alone, err := followUpChoice(ctx, managed, decisionFollowUp)
if err != nil {
return err
}
withPrior, err := followUpChoice(ctx, managed, first+"\n\n"+decisionFollowUp)
if err != nil {
return err
}
if alone == withPrior {
continue
}
messages := []map[string]string{
{"role": "user", "content": first},
{"role": "assistant", "content": "Here it is."},
{"role": "user", "content": decisionFollowUp},
}
response, err := sendLocalChatConversation(ctx, session.gatewayPort, "vllm-sr/auto", messages, mrRequestTimeout)
if err != nil {
return err
}
if response.StatusCode != http.StatusOK {
return fmt.Errorf("follow-up chat: %s", formatUnexpectedChatCompletionStatus(response))
}
matched := headerItems(response.Headers, "x-vsr-matched-decision-model")
if !matched["follow_up_kind:"+withPrior] || matched["follow_up_kind:"+alone] {
return fmt.Errorf("x-vsr-matched-decision-model %v: want follow_up_kind:%s (the runtime's answer after %q), not follow_up_kind:%s (the follow-up alone)",
matched, withPrior, first, alone)
}
if opts.SetDetails != nil {
opts.SetDetails(map[string]interface{}{"first_turn": first, "alone": alone, "with_prior_turn": withPrior})
}
return nil
}
return fmt.Errorf("the fixture answers %q the same alone and after each of %d first turns, so no route can show which state was asked", decisionFollowUp, len(decisionFirstTurns))
}

func followUpChoice(ctx context.Context, runtime *modelruntime.Client, state string) (string, error) {
answers, err := runtime.Decide(ctx, modelruntime.DecisionsRequest{Model: mrDecisionDeployment, State: state, Questions: decisionFollowUpQuestion})
if err != nil {
return "", err
}
return answers.Answers["follow_up_kind"].Choice, nil
}
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package classification
import (
"context"
"slices"
"strings"
"sync"
"time"

Expand All @@ -27,46 +28,71 @@ func (c *Classifier) decider() modelservice.Decider {
return modelservice.Default()
}

// decisionCall is one call's deployment and the earlier user turns its
// questions read.
type decisionCall struct {
deployment string
priorTurns int
}

// evaluateDecisionModelSignals asks every used decision question in one call
// per deployment, with the request text as the state. Calls to different
// deployments run in parallel. A question that names no deployment asks the
// decision model, with the request as it came, like the built-in signals
// whose call it joins. A failed or late call leaves its signals unknown; it
// never fails the request.
// whose call it joins. Questions that read earlier user turns share a call
// only with questions that read as many. A failed or late call leaves its
// signals unknown; it never fails the request.
func (c *Classifier) evaluateDecisionModelSignals(
ctx context.Context,
results *SignalResults,
mu *sync.Mutex,
text string,
wholeText string,
priorUserMessages []string,
usedSignals map[string]bool,
) {
if ctx == nil {
ctx = context.Background()
}
start := time.Now()
byDeployment := make(map[string][]config.DecisionSignalRule)
stateOf := make(map[string]string)
var order []string
byCall := make(map[decisionCall][]config.DecisionSignalRule)
stateOf := make(map[decisionCall]string)
var order []decisionCall
for _, rule := range c.Config.DecisionRules {
if !signalRuleUsed(usedSignals, config.SignalTypeDecision, rule.Name) {
continue
}
deployment := c.Config.DecisionQuestionDeployment(rule)
if _, seen := byDeployment[deployment]; !seen {
order = append(order, deployment)
stateOf[deployment] = wholeText
call := decisionCall{c.Config.DecisionQuestionDeployment(rule), min(rule.PriorUserTurns, len(priorUserMessages))}
if _, seen := byCall[call]; !seen {
order = append(order, call)
stateOf[call] = withPriorUserTurns(wholeText, priorUserMessages, call.priorTurns)
}
byDeployment[deployment] = append(byDeployment[deployment], rule)
byCall[call] = append(byCall[call], rule)
}
modelservice.Fan(ctx, len(order), func(i int) {
c.evaluateDecisionDeployment(ctx, results, mu, stateOf[order[i]], order[i], byDeployment[order[i]])
c.evaluateDecisionDeployment(ctx, results, mu, stateOf[order[i]], order[i].deployment, byCall[order[i]])
})
mu.Lock()
results.Metrics.Decision.ExecutionTimeMs = float64(time.Since(start).Microseconds()) / 1000.0
mu.Unlock()
}

// withPriorUserTurns is state after the last turns of prior, oldest first,
// one blank line apart. The earlier turns share semanticSignalUnitLimit runes,
// the latest turn first and each cut to its first runes, so however many turns
// a question reads they add the same bounded context to the current turn.
func withPriorUserTurns(state string, prior []string, turns int) string {
parts := []string{state}
room := semanticSignalUnitLimit
for index := len(prior) - 1; index >= len(prior)-turns && room > 0; index-- {
runes := []rune(prior[index])
runes = runes[:min(len(runes), room)]
room -= len(runes)
parts = append([]string{string(runes)}, parts...)
}
return strings.Join(parts, "\n\n")
}

func (c *Classifier) evaluateDecisionDeployment(
ctx context.Context,
results *SignalResults,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package classification
import (
"context"
"slices"
"strings"
"sync"
"testing"

Expand Down Expand Up @@ -84,7 +85,7 @@ func TestDecisionSignalsBatchPerDeploymentAndMatch(t *testing.T) {
for _, rule := range decisionRulesForTest() {
used["decision:"+rule.Name] = true
}
classifier.evaluateDecisionModelSignals(context.Background(), results, &sync.Mutex{}, "merge two lists", "merge two lists", used)
classifier.evaluateDecisionModelSignals(context.Background(), results, &sync.Mutex{}, "merge two lists", "merge two lists", nil, used)

if decider.calls["kai"] != 1 || decider.calls["vega"] != 1 {
t.Fatalf("one call per deployment expected, got %v", decider.calls)
Expand Down Expand Up @@ -169,7 +170,7 @@ func TestSetAndSpanAnswersMatchLabelsAndPublishEveryProbability(t *testing.T) {
func TestUnusedDecisionSignalsAreNotAsked(t *testing.T) {
decider := &fakeDecider{calls: map[string]int{}, requests: map[string]modelservice.Request{}, answers: map[string]modelservice.Answer{}}
classifier := newDecisionTestClassifier(decider)
classifier.evaluateDecisionModelSignals(context.Background(), newSignalResults(), &sync.Mutex{}, "x", "x", map[string]bool{"decision:remote": true})
classifier.evaluateDecisionModelSignals(context.Background(), newSignalResults(), &sync.Mutex{}, "x", "x", nil, map[string]bool{"decision:remote": true})
if decider.calls["kai"] != 0 || decider.calls["vega"] != 1 {
t.Fatalf("only used signals are asked: %v", decider.calls)
}
Expand Down Expand Up @@ -201,7 +202,7 @@ func TestADecisionQuestionWithoutDeploymentJoinsTheDecisionModelsCall(t *testing
classifier := &Classifier{Config: cfg}
classifier.SetDecisionDecider(decider)
used := map[string]bool{"decision:tools": true, "decision:hard": true}
classifier.evaluateDecisionModelSignals(context.Background(), newSignalResults(), &sync.Mutex{}, "short", "the whole text", used)
classifier.evaluateDecisionModelSignals(context.Background(), newSignalResults(), &sync.Mutex{}, "short", "the whole text", nil, used)

if decider.states[config.DefaultDecisionDeployment] != "the whole text" || decider.states["kai"] != "the whole text" {
t.Fatalf("every decision question reads the same authored input: %v", decider.states)
Expand All @@ -210,3 +211,74 @@ func TestADecisionQuestionWithoutDeploymentJoinsTheDecisionModelsCall(t *testing
t.Fatalf("only a declared deployment's call takes the default timeout: %v", decider.deadlines)
}
}

type recordingDecider struct {
mu sync.Mutex
requests []modelservice.Request
}

func (r *recordingDecider) Decide(_ context.Context, _ string, request modelservice.Request) (modelservice.Response, error) {
r.mu.Lock()
defer r.mu.Unlock()
r.requests = append(r.requests, request)
return modelservice.Response{Answers: map[string]modelservice.Answer{}}, nil
}

// A follow-up such as "now make it shorter" says nothing about its task;
// a question with prior_user_turns reads it after the turns it follows,
// and only questions that read as many turns share its call.
func TestDecisionQuestionsReadTheirPriorUserTurns(t *testing.T) {
decider := &recordingDecider{}
cfg := &config.RouterConfig{}
cfg.DecisionRules = []config.DecisionSignalRule{
{Name: "alone", Deployment: "kai", Question: config.DecisionQuestion{Type: "noul", Instructions: "?"}},
{Name: "one", Deployment: "kai", PriorUserTurns: 1, Question: config.DecisionQuestion{Type: "noul", Instructions: "?"}},
{Name: "many", Deployment: "kai", PriorUserTurns: 5, Question: config.DecisionQuestion{Type: "noul", Instructions: "?"}},
}
classifier := &Classifier{Config: cfg}
classifier.SetDecisionDecider(decider)
used := map[string]bool{"decision:alone": true, "decision:one": true, "decision:many": true}
latest := strings.Repeat("ü", semanticSignalUnitLimit-10)
prior := []string{"Write a poem about rain.", latest}

classifier.evaluateDecisionModelSignals(context.Background(), newSignalResults(), &sync.Mutex{}, "Now make it shorter.", "Now make it shorter.", prior, used)

states := map[string]string{}
for _, request := range decider.requests {
for _, question := range request.Questions {
states[question.ID] = request.State
}
}
want := map[string]string{
"alone": "Now make it shorter.",
"one": latest + "\n\nNow make it shorter.",
// five asked, two exist: oldest first, within one shared budget
"many": "Write a po\n\n" + latest + "\n\nNow make it shorter.",
}
for name, state := range want {
if states[name] != state {
t.Errorf("%s: state %q, want %q", name, states[name], state)
}
}
if len(decider.requests) != 3 {
t.Fatalf("questions reading different turns need separate calls, got %d", len(decider.requests))
}
}

func TestPriorUserTurnsWithoutHistoryShareTheCurrentTurnsCall(t *testing.T) {
decider := &recordingDecider{}
cfg := &config.RouterConfig{}
cfg.DecisionRules = []config.DecisionSignalRule{
{Name: "alone", Deployment: "kai", Question: config.DecisionQuestion{Type: "noul", Instructions: "?"}},
{Name: "one", Deployment: "kai", PriorUserTurns: 1, Question: config.DecisionQuestion{Type: "noul", Instructions: "?"}},
}
classifier := &Classifier{Config: cfg}
classifier.SetDecisionDecider(decider)
used := map[string]bool{"decision:alone": true, "decision:one": true}

classifier.evaluateDecisionModelSignals(context.Background(), newSignalResults(), &sync.Mutex{}, "hello", "hello", nil, used)

if len(decider.requests) != 1 || decider.requests[0].State != "hello" || len(decider.requests[0].Questions) != 2 {
t.Fatalf("a first turn has no earlier turns, so both questions share one call: %+v", decider.requests)
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ func (c *Classifier) buildPolicySignalDispatchers(
mu,
textForSignal(config.SignalTypeDecision),
textForSignal(decisionModelQuestionText),
priorUserMessages,
usedSignals,
)
},
Expand Down
Loading
Loading