From beb8e516a4b85407c54c26d33cebd8589251198e Mon Sep 17 00:00:00 2001 From: Samuel Abolo Date: Thu, 29 Oct 2026 00:32:04 +0300 Subject: [PATCH 1/2] Score on 5 key questions --- README.md | 75 +++++++++++++++++++++++---- examples/manifests/my-agent.json | 33 +++++++++++- internal/config/builtin_manifest.json | 33 +++++++++++- internal/config/config.go | 9 ++++ internal/config/config_test.go | 38 ++++++++++++++ internal/config/engine.go | 1 + internal/config/manifest.go | 35 +++++++++++++ internal/pipeline/engine.go | 5 +- internal/pipeline/pool.go | 40 ++++++++++++++ internal/pipeline/pool_test.go | 36 +++++++++++++ internal/pipeline/types.go | 9 +++- internal/scorer/scorer.go | 26 +++++++++- internal/scorer/scorer_test.go | 34 ++++++++++++ internal/server/server_test.go | 10 +++- 14 files changed, 369 insertions(+), 15 deletions(-) create mode 100644 internal/pipeline/pool.go create mode 100644 internal/pipeline/pool_test.go diff --git a/README.md b/README.md index 40c276a..a855fb5 100644 --- a/README.md +++ b/README.md @@ -262,18 +262,69 @@ A manifest holds the questions the model is asked: ``` - `questions` is sent to the model unchanged, all questions, in the - manifest's order. The decision question's answer is `p_keep`, and the - reason question's chosen option is the `reason`. + manifest's order, in one call. The reason question's chosen option is the + `reason`. +- `decision`, optional, says how the yes/no (`noul`) answers make `p_keep` + (see [Combining the answers](#combining-the-answers)). Without it, `p_keep` + is the decision question's answer. It stays in the engine and is not sent + to the model. - A relative `scorer.manifest` path resolves against the config file's directory. A manifest is optional, for every System One model. Without one the model -is asked the built-in questions (`answer_wrong`, and `why_wrong` for the -reason), and the engine logs a warning at start-up. To word the questions +is asked the built-in questions +([`builtin_manifest.json`](internal/config/builtin_manifest.json)), and the +engine logs a warning at start-up. They are: + +| Question | Weight | Asks | +| --- | --- | --- | +| `answer_wrong` | 0.5 | Is the final answer wrong or only partially correct? | +| `unsupported_answer` | 0.1 | Does it make claims the evidence does not support? | +| `missing_tool_calls` | 0.1 | Did the agent skip lookups the task needed? | +| `answered_on_failure` | 0.1 | Did it answer despite a failed or partial tool call? | +| `ignored_instructions` | 0.1 | Does it go against the user's instructions? | +| `contradicts_knowledge` | 0.1 | Does it contradict general knowledge? | +| `why_wrong` | | Which [reason](#reasons) explains what is wrong? | + +Each yes/no answer is the probability that the agent failed in that way, and +the weights combine them into `p_keep`. To word the questions for your agent, start from [`examples/manifests/my-agent.json`](examples/manifests/my-agent.json) and set `scorer.manifest` (or `TAILWISE_MANIFEST`). +### Combining the answers + +A manifest's `decision` block weights its yes/no questions: + +```json +"decision": { + "intercept": 0, + "weights": {"answer_wrong": 0.5, "unsupported_answer": 0.1, "missing_tool_calls": 0.1, + "answered_on_failure": 0.1, "ignored_instructions": 0.1, "contradicts_knowledge": 0.1} +} +``` + +and `p_keep` is a weighted [logarithmic opinion pool](https://en.wikipedia.org/wiki/Logarithmic_opinion_pool) +of their answers: + +``` +p_keep = sigmoid(intercept + Σ weight × logit(answer)) logit(p) = ln(p / (1 − p)) +``` + +- **Weights summing to 1, no intercept** (as in the built-in questions): + `p_keep` stays on the answers' own scale. Answers that agree keep their + probability (all at 0.9 give 0.9); answers that disagree meet in between. + A larger weight gives a question more say. +- **Fitted weights.** The formula is a logistic regression on the answers' + logits, so the coefficients and intercept of one fitted on labelled traces + can be used as they are. They need not sum to 1. +- **Rules.** Every weighted question must be a `noul` question of the same + manifest, and the model must answer all of them: a missing answer is a + scoring error, like a missing decision answer. Questions without a weight + are still asked and logged, but do not change `p_keep`. +- **Changing the weights changes `p_keep`,** so recheck `policy.keep_threshold` + against the share of traces kept. + ### One model per agent Each agent can have its own model: its own questions, and @@ -566,12 +617,16 @@ so it does not take the engine out of a load balancer; watch queue for redelivery. On the synchronous endpoint, which has no queue, the engine retries it with `scorer.max_retries` and jittered `scorer.backoff`. The two kinds of retry never multiply. -4. **Decide.** `p_keep` is the model's answer to the decision question. +4. **Decide.** `p_keep` combines the model's yes/no answers with the + manifest's weights (see [Combining the answers](#combining-the-answers)), + or, for a manifest without them, is its answer to the decision question. - `p_keep >= keep_threshold`: keep, rule `above_threshold`. - Otherwise keep with probability `random_floor`, rule `random_floor`. - Otherwise drop, rule `below_threshold`. 5. **Export.** Kept traces are queued to every exporter. Each decision produces one JSON log - line with `trace_id`, `agent`, `decision`, `rule`, `reason`, `cause`, `p_keep`, `model` and `latency_ms`. + line with `trace_id`, `agent`, `decision`, `rule`, `reason`, `cause`, `p_keep`, `model`, `latency_ms`, + and `answers`: every yes/no answer, by question, as in + `"answers": {"answer_wrong": 0.91, "unsupported_answer": 0.84, ...}`. ### Reasons @@ -949,9 +1004,11 @@ These are open problems in the current version, not design choices to rely on. a change to its request schema), every trace is decided at random, about 98% dropped with the default `random_floor`, and only `WARN` log lines show it. -- **Decisions use the model's probabilities as they come.** Nothing tunes - them to your agent, so `keep_threshold` may keep too much or too little at - first. Watch the share of traces kept and adjust the threshold. +- **The built-in weights are a reasonable guess, not fitted.** Nothing tunes + them, or the model's probabilities, to your agent, so `keep_threshold` may + keep too much or too little at first. Watch the share of traces kept and + adjust the threshold; with labelled traces, fit the weights (see + [Combining the answers](#combining-the-answers)). - **Every scored trace is a billed call to Jev, and leaves your network.** Nothing caps the spend or redacts fields before sending; only `server.workers` limits the rate. diff --git a/examples/manifests/my-agent.json b/examples/manifests/my-agent.json index 4e0dc36..b1aef32 100644 --- a/examples/manifests/my-agent.json +++ b/examples/manifests/my-agent.json @@ -1,11 +1,31 @@ { "model": "jev-latest", - "note": "A manifest for your agent: the questions it is asked. answer_wrong decides keep or drop, and why_wrong picks the plain-English reason; word both for your agent.", + "note": "A manifest for your agent: the questions it is asked, all in one call. answer_wrong and the five checks after it are combined into p_keep with the weights in decision, and why_wrong picks the plain-English reason; word them all for your agent, and weight what matters to it.", "questions": { "answer_wrong": { "type": "noul", "instructions": "Is the final answer wrong or only partially correct for the user's question?" }, + "unsupported_answer": { + "type": "noul", + "instructions": "Does the final answer make claims that the evidence the agent gathered does not support?" + }, + "missing_tool_calls": { + "type": "noul", + "instructions": "Did the agent leave out tool calls the task needed, so it answered without information it should have looked up?" + }, + "answered_on_failure": { + "type": "noul", + "instructions": "Did the agent answer even though a tool call failed or returned only partial results?" + }, + "ignored_instructions": { + "type": "noul", + "instructions": "Does the final answer ignore or go against instructions given in the user's question?" + }, + "contradicts_knowledge": { + "type": "noul", + "instructions": "Does the final answer contradict well-established general knowledge?" + }, "why_wrong": { "type": "choice", "instructions": "Which best explains what is wrong with the final answer? Choose \"correct\" if nothing is wrong.", @@ -19,5 +39,16 @@ "no_answer": "The agent did not give an answer." } } + }, + "decision": { + "intercept": 0, + "weights": { + "answer_wrong": 0.5, + "unsupported_answer": 0.1, + "missing_tool_calls": 0.1, + "answered_on_failure": 0.1, + "ignored_instructions": 0.1, + "contradicts_knowledge": 0.1 + } } } diff --git a/internal/config/builtin_manifest.json b/internal/config/builtin_manifest.json index 34f301a..fc08878 100644 --- a/internal/config/builtin_manifest.json +++ b/internal/config/builtin_manifest.json @@ -1,11 +1,31 @@ { "model": "built-in", - "note": "The questions the engine asks when a model has no manifest of its own: answer_wrong decides keep or drop, and why_wrong picks the plain-English reason.", + "note": "The questions the engine asks when a model has no manifest of its own, all in one call: answer_wrong and the five checks after it are combined into p_keep with the weights in decision, and why_wrong picks the plain-English reason.", "questions": { "answer_wrong": { "type": "noul", "instructions": "Is the final answer wrong or only partially correct for the user's question?" }, + "unsupported_answer": { + "type": "noul", + "instructions": "Does the final answer make claims that the evidence the agent gathered does not support?" + }, + "missing_tool_calls": { + "type": "noul", + "instructions": "Did the agent leave out tool calls the task needed, so it answered without information it should have looked up?" + }, + "answered_on_failure": { + "type": "noul", + "instructions": "Did the agent answer even though a tool call failed or returned only partial results?" + }, + "ignored_instructions": { + "type": "noul", + "instructions": "Does the final answer ignore or go against instructions given in the user's question?" + }, + "contradicts_knowledge": { + "type": "noul", + "instructions": "Does the final answer contradict well-established general knowledge?" + }, "why_wrong": { "type": "choice", "instructions": "Which best explains what is wrong with the final answer? Choose \"correct\" if nothing is wrong.", @@ -19,5 +39,16 @@ "no_answer": "The agent did not give an answer." } } + }, + "decision": { + "intercept": 0, + "weights": { + "answer_wrong": 0.5, + "unsupported_answer": 0.1, + "missing_tool_calls": 0.1, + "answered_on_failure": 0.1, + "ignored_instructions": 0.1, + "contradicts_knowledge": 0.1 + } } } diff --git a/internal/config/config.go b/internal/config/config.go index fd789f2..2a64402 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -181,10 +181,19 @@ type Manifest struct { Model string `json:"model"` // Questions is kept as written (compacted), so keys reach the model in order. Questions json.RawMessage `json:"questions"` + // Decision, when set, combines the yes/no answers into p_keep; it stays in + // the engine and is not sent to the model. + Decision *ManifestDecision `json:"decision"` Reasons map[string]string `json:"-"` } +// ManifestDecision weights each yes/no question's answer; see pipeline.Pool. +type ManifestDecision struct { + Intercept float64 `json:"intercept"` + Weights map[string]float64 `json:"weights"` +} + // DefaultModel is what an agent's model starts from: the built-in scorer // settings, minus the endpoint and model it must set itself. func DefaultModel() Scorer { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 17ecfd7..9c40a18 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -277,6 +277,10 @@ func TestLoadWithoutAFile(t *testing.T) { if m.Client.ReasonQuestion != "why_wrong" || m.Client.Reasons["contradicts_evidence"] != "The final answer contradicts the evidence the agent gathered." { t.Errorf("reason question %q, reasons %v", m.Client.ReasonQuestion, m.Client.Reasons) } + // They combine answer_wrong and the five checks into p_keep. + if p := m.Client.Pool; p == nil || len(p.Weights) != 6 || p.Weights["answer_wrong"] != 0.5 || p.Weights["contradicts_knowledge"] != 0.1 { + t.Errorf("pool = %+v, want the built-in weights", p) + } t.Setenv(DefaultJevAPIKeyEnv, "") if _, err := Load(""); err == nil || !strings.Contains(err.Error(), "$JEV_API_KEY is not set") { t.Errorf("no key: error = %v", err) @@ -380,3 +384,37 @@ func TestNATSNamesFromThePrefix(t *testing.T) { t.Errorf("names = %q\nwant %q", got, want) } } + +// A manifest's decision block becomes the scorer's pool; it may weight only +// its own yes/no questions. Without one, p_keep is the decision answer. +func TestManifestDecisionWeights(t *testing.T) { + cfg, err := Load(write(t, "scorer:\n manifest: manifest.json\n", testManifest)) + if err != nil || cfg.Models.Default.Client.Pool != nil { + t.Fatalf("no decision block: pool %+v, error %v", cfg.Models.Default.Client.Pool, err) + } + + with := func(decision string) string { + return `{"questions": {"answer_wrong": {"type": "noul"}, "checks": {"type": "noul"},` + + ` "quality": {"type": "score", "criteria": ["a", "b"]}}, "decision": ` + decision + `}` + } + cfg, err = Load(write(t, "scorer:\n manifest: manifest.json\n", with(`{"intercept": -1, "weights": {"answer_wrong": 2, "checks": 0.5}}`))) + if err != nil { + t.Fatal(err) + } + if p := cfg.Models.Default.Client.Pool; p == nil || p.Intercept != -1 || p.Weights["answer_wrong"] != 2 || p.Weights["checks"] != 0.5 { + t.Errorf("pool = %+v", p) + } + if strings.Contains(string(cfg.Models.Default.Client.Questions), "weights") { + t.Errorf("the decision block reached the questions sent to the model: %s", cfg.Models.Default.Client.Questions) + } + + for decision, wantErr := range map[string]string{ + `{"weights": {}}`: "must weight at least one question", + `{"weights": {"tone": 1}}`: `question "tone" not found`, + `{"weights": {"quality": 1}}`: `question "quality" has type "score", want "noul"`, + } { + if _, err := Load(write(t, "scorer:\n manifest: manifest.json\n", with(decision))); err == nil || !strings.Contains(err.Error(), wantErr) { + t.Errorf("%s: error = %v, want %q", decision, err, wantErr) + } + } +} diff --git a/internal/config/engine.go b/internal/config/engine.go index 9cd6107..6af738c 100644 --- a/internal/config/engine.go +++ b/internal/config/engine.go @@ -140,6 +140,7 @@ func model(s Scorer, apiKey string, m *Manifest) scorer.ModelConfig { Model: s.Model, Questions: m.Questions, DecisionQuestion: s.DecisionQuestion, + Pool: m.pool(), ReasonQuestion: reasonQuestion(s, m), Reasons: m.Reasons, APIKey: apiKey, diff --git a/internal/config/manifest.go b/internal/config/manifest.go index 636ae67..37267c0 100644 --- a/internal/config/manifest.go +++ b/internal/config/manifest.go @@ -6,7 +6,10 @@ import ( _ "embed" "encoding/json" "fmt" + "maps" "os" + + "github.com/tecnosam/tailwise/internal/pipeline" ) //go:embed builtin_manifest.json @@ -99,5 +102,37 @@ func (m *Manifest) prepare(decisionQuestion string) error { if head.Type != questionTypeNoul { return fmt.Errorf("decision question %q has type %q, want %q", decisionQuestion, head.Type, questionTypeNoul) } + return m.checkDecision(questions) +} + +// checkDecision accepts a decision block whose weights name only yes/no +// questions of the manifest. +func (m *Manifest) checkDecision(questions map[string]json.RawMessage) error { + if m.Decision == nil { + return nil + } + if len(m.Decision.Weights) == 0 { + return fmt.Errorf("\"decision.weights\" must weight at least one question") + } + for q := range m.Decision.Weights { + raw, ok := questions[q] + if !ok { + return fmt.Errorf("decision.weights: question %q not found in questions", q) + } + var head struct { + Type string `json:"type"` + } + if err := json.Unmarshal(raw, &head); err != nil || head.Type != questionTypeNoul { + return fmt.Errorf("decision.weights: question %q has type %q, want %q", q, head.Type, questionTypeNoul) + } + } return nil } + +// pool is the decision block as the scorer uses it, nil without one. +func (m *Manifest) pool() *pipeline.Pool { + if m.Decision == nil { + return nil + } + return &pipeline.Pool{Intercept: m.Decision.Intercept, Weights: maps.Clone(m.Decision.Weights)} +} diff --git a/internal/pipeline/engine.go b/internal/pipeline/engine.go index 52d83ef..447f2d0 100644 --- a/internal/pipeline/engine.go +++ b/internal/pipeline/engine.go @@ -110,7 +110,7 @@ func (e *Engine) scored(t trace.Trace, model Model, score Score) Record { if d.Reason == "" { d.Reason = probabilityReason(score.P) } - return Record{Trace: t, Agent: model.Agent, Decision: d, Model: score.Model, Latency: score.Latency} + return Record{Trace: t, Agent: model.Agent, Decision: d, Model: score.Model, Latency: score.Latency, Answers: score.Answers} } // probabilityReason is the reason for a model that gave none of its own. @@ -162,6 +162,9 @@ func (e *Engine) logDecision(out Outcome) { if d.Scored { attrs = append(attrs, "p_keep", float64(d.PKeep)) } + if len(rec.Answers) > 0 { + attrs = append(attrs, "answers", rec.Answers) + } e.deps.Logger.Info("decision", attrs...) } diff --git a/internal/pipeline/pool.go b/internal/pipeline/pool.go new file mode 100644 index 0000000..d307aff --- /dev/null +++ b/internal/pipeline/pool.go @@ -0,0 +1,40 @@ +package pipeline + +import ( + "fmt" + "maps" + "math" + "slices" +) + +// Pool combines the model's answers into p_keep as a weighted logarithmic +// opinion pool: +// +// p_keep = sigmoid(Intercept + Σ Weights[q] · logit(answer q)) +// +// With weights summing to 1 and no intercept, answers that agree keep their +// probability and answers that disagree meet in between. A logistic +// regression fitted on the answers' logits has the same form, so its +// coefficients can be used as the weights. +type Pool struct { + Intercept float64 + Weights map[string]float64 +} + +// poolEpsilon keeps the logit of an answer of exactly 0 or 1 finite. +const poolEpsilon = 1e-6 + +// Combine needs an answer for every weighted question. Questions are summed +// in name order, so the same answers always give the same p_keep. +func (p *Pool) Combine(answers map[string]Probability) (Probability, error) { + z := p.Intercept + for _, q := range slices.Sorted(maps.Keys(p.Weights)) { + a, ok := answers[q] + if !ok { + return 0, fmt.Errorf("model response has no answer for %q", q) + } + x := min(max(float64(a), poolEpsilon), 1-poolEpsilon) + z += p.Weights[q] * math.Log(x/(1-x)) + } + return Probability(1 / (1 + math.Exp(-z))), nil +} diff --git a/internal/pipeline/pool_test.go b/internal/pipeline/pool_test.go new file mode 100644 index 0000000..3d0af57 --- /dev/null +++ b/internal/pipeline/pool_test.go @@ -0,0 +1,36 @@ +package pipeline + +import ( + "math" + "strings" + "testing" +) + +func TestPoolCombines(t *testing.T) { + half := &Pool{Weights: map[string]float64{"a": 0.5, "b": 0.5}} + for _, tt := range []struct { + name string + pool *Pool + answers map[string]Probability + want float64 + }{ + {"agreeing answers keep their probability", half, map[string]Probability{"a": 0.9, "b": 0.9}, 0.9}, + {"opposite answers meet in the middle", half, map[string]Probability{"a": 0.9, "b": 0.1}, 0.5}, + {"one question with weight 1 is its answer", &Pool{Weights: map[string]float64{"a": 1}}, map[string]Probability{"a": 0.3}, 0.3}, + {"the intercept shifts the log-odds", &Pool{Intercept: math.Log(9), Weights: map[string]float64{"a": 1}}, map[string]Probability{"a": 0.5}, 0.9}, + {"unweighted answers are ignored", half, map[string]Probability{"a": 0.2, "b": 0.2, "c": 1}, 0.2}, + {"an answer of exactly 1 stays finite", half, map[string]Probability{"a": 1, "b": 0.5}, 0.999}, + } { + got, err := tt.pool.Combine(tt.answers) + if err != nil || math.Abs(float64(got)-tt.want) > 1e-3 { + t.Errorf("%s: got %v, %v; want %v", tt.name, got, err, tt.want) + } + } +} + +func TestPoolNeedsEveryWeightedAnswer(t *testing.T) { + p := &Pool{Weights: map[string]float64{"a": 0.5, "b": 0.5}} + if _, err := p.Combine(map[string]Probability{"a": 0.9}); err == nil || !strings.Contains(err.Error(), `no answer for "b"`) { + t.Errorf("error = %v", err) + } +} diff --git a/internal/pipeline/types.go b/internal/pipeline/types.go index 8dee613..77b88ea 100644 --- a/internal/pipeline/types.go +++ b/internal/pipeline/types.go @@ -24,13 +24,17 @@ const ( ) type Score struct { - P Probability // that the final answer is wrong + // P is p_keep: the decision question's answer, or the manifest's pool of + // answers. + P Probability Model ModelName Latency time.Duration // as the server reported it, else the round trip // Cause is the model's choice for the reason question, Explanation its // plain-English description; both empty when it gave none. Cause string Explanation string + // Answers holds every yes/no (noul) answer the model gave, by question. + Answers map[string]Probability } type Decision struct { @@ -53,6 +57,9 @@ type Record struct { Decision Decision Model ModelName Latency time.Duration + // Answers are the model's yes/no answers p_keep was decided from; none on + // fallback. + Answers map[string]Probability } type Outcome struct { diff --git a/internal/scorer/scorer.go b/internal/scorer/scorer.go index 7f138c7..fb00108 100644 --- a/internal/scorer/scorer.go +++ b/internal/scorer/scorer.go @@ -30,6 +30,9 @@ type Config struct { Model string Questions json.RawMessage DecisionQuestion string + // Pool, when set, combines the yes/no answers into p_keep; without it + // p_keep is the decision question's answer. + Pool *pipeline.Pool // Reasons maps each option of the reason question to the plain-English // sentence that becomes the decision's reason. ReasonQuestion string @@ -158,7 +161,13 @@ func (c *Client) parse(raw []byte, elapsed time.Duration) (pipeline.Score, error if err != nil { return pipeline.Score{}, err } - score := pipeline.Score{P: p, Model: c.modelName(out), Latency: reportedLatency(out, elapsed)} + answers := noulAnswers(out) + if c.cfg.Pool != nil { + if p, err = c.cfg.Pool.Combine(answers); err != nil { + return pipeline.Score{}, err + } + } + score := pipeline.Score{P: p, Answers: answers, Model: c.modelName(out), Latency: reportedLatency(out, elapsed)} score.Cause, score.Explanation = c.reasonAnswer(out) return score, nil } @@ -202,6 +211,21 @@ func (c *Client) decisionAnswer(out response) (pipeline.Probability, error) { return pipeline.Probability(*ans.Noul), nil } +// noulAnswers is every well-formed yes/no answer; the rest are left out. +func noulAnswers(out response) map[string]pipeline.Probability { + answers := make(map[string]pipeline.Probability, len(out.Answers)) + for q, raw := range out.Answers { + var ans noulAnswer + if json.Unmarshal(raw, &ans) != nil || ans.Noul == nil { + continue + } + if p := *ans.Noul; !math.IsNaN(p) && p >= 0 && p <= 1 { + answers[q] = pipeline.Probability(p) + } + } + return answers +} + func (c *Client) modelName(out response) pipeline.ModelName { if out.Model != "" { return pipeline.ModelName(out.Model) diff --git a/internal/scorer/scorer_test.go b/internal/scorer/scorer_test.go index 251ccae..47e5f25 100644 --- a/internal/scorer/scorer_test.go +++ b/internal/scorer/scorer_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "io" + "math" "net/http" "net/http/httptest" "os" @@ -15,6 +16,7 @@ import ( "time" "github.com/tecnosam/tailwise/internal/fault" + "github.com/tecnosam/tailwise/internal/pipeline" ) const questions = `{"answer_wrong":{"type":"noul","instructions":"Is the final answer wrong?"},"answers_question":{"type":"score","instructions":"How good?","criteria":["Wrong","Fully correct"]}}` @@ -378,3 +380,35 @@ func TestScoreReadsTheReason(t *testing.T) { } } } + +// Every yes/no answer is kept; with a pool, p_keep combines them, and without +// one it is the decision question's answer. +func TestScoreKeepsAndPoolsTheAnswers(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"answers": {"answer_wrong": {"type": "noul", "noul": 0.9}, "unsupported_answer": {"type": "noul", "noul": 0.1},` + + ` "answers_question": {"type": "score", "score": 1.2}, "bad": {"type": "noul", "noul": 3}}}`)) + })) + defer srv.Close() + + res, err := newClient(t, srv.URL, nil).Score(context.Background(), "s") + if err != nil || res.P != 0.9 { + t.Fatalf("without a pool: %+v, %v", res, err) + } + if len(res.Answers) != 2 || res.Answers["answer_wrong"] != 0.9 || res.Answers["unsupported_answer"] != 0.1 { + t.Errorf("answers = %v, want the two well-formed yes/no answers", res.Answers) + } + + pooled := newClient(t, srv.URL, func(cfg *Config) { + cfg.Pool = &pipeline.Pool{Weights: map[string]float64{"answer_wrong": 0.5, "unsupported_answer": 0.5}} + }) + if res, err = pooled.Score(context.Background(), "s"); err != nil || math.Abs(float64(res.P)-0.5) > 1e-9 { + t.Errorf("with a pool: %+v, %v; want p_keep 0.5", res, err) + } + + missing := newClient(t, srv.URL, func(cfg *Config) { + cfg.Pool = &pipeline.Pool{Weights: map[string]float64{"answer_wrong": 0.5, "missing_tool_calls": 0.5}} + }) + if _, err = missing.Score(context.Background(), "s"); err == nil || !strings.Contains(err.Error(), `no answer for "missing_tool_calls"`) { + t.Errorf("a weighted question without an answer: error = %v", err) + } +} diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 12f975a..023220b 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -655,8 +655,12 @@ func TestScoreDoesNotWaitForTheBackEnd(t *testing.T) { // One JSON log line per decision with the fields operators filter on. func TestDecisionLogLine(t *testing.T) { var buf strings.Builder + answered := func(context.Context, trace.State) (pipeline.Score, error) { + return pipeline.Score{P: 0.9, Model: "kev-latest", Latency: 12 * time.Millisecond, + Answers: map[string]pipeline.Probability{"answer_wrong": 0.95, "unsupported_answer": 0.7}}, nil + } h := newHarness(t, harnessOpts{ - scorer: &fakeScorer{score: fixedScore(0.9)}, + scorer: &fakeScorer{score: answered}, logger: jsonLogger(&buf), }) h.do("POST", "/v1/traces:score", `{"trace_id": "log-1", "question": "q", "final_answer": "a"}`) @@ -675,6 +679,10 @@ func TestDecisionLogLine(t *testing.T) { t.Errorf("log field %s = %v, want %v", k, line[k], v) } } + answers, _ := line["answers"].(map[string]any) + if answers["answer_wrong"] != 0.95 || answers["unsupported_answer"] != 0.7 { + t.Errorf("log field answers = %v, want every yes/no answer", line["answers"]) + } } func deref(p *float64) any { From 7ff0256ce920a26a5aeef842eff4a00d6a8932be Mon Sep 17 00:00:00 2001 From: Samuel Abolo Date: Thu, 29 Oct 2026 13:41:05 +0300 Subject: [PATCH 2/2] feat(decision): keep a trace when any check flags it, with per-question thresholds and severity --- README.md | 123 ++++++++++++-------- architecture/sequence-diagrams.md | 2 +- examples/manifests/my-agent.json | 19 +-- internal/config/builtin_manifest.json | 18 ++- internal/config/config.go | 14 ++- internal/config/config_test.go | 66 ++++++++--- internal/config/engine.go | 5 +- internal/config/manifest.go | 54 ++++++--- internal/export/natsq/queue.go | 11 +- internal/export/natsq/queue_test.go | 9 +- internal/export/otlpbackend/backend.go | 9 ++ internal/export/otlpbackend/backend_test.go | 13 ++- internal/pipeline/checks.go | 25 ++++ internal/pipeline/engine.go | 8 +- internal/pipeline/model.go | 3 + internal/pipeline/policy.go | 28 ++++- internal/pipeline/policy_test.go | 65 +++++++++++ internal/pipeline/pool.go | 40 ------- internal/pipeline/pool_test.go | 36 ------ internal/pipeline/types.go | 16 ++- internal/scorer/factory.go | 4 +- internal/scorer/scorer.go | 18 +-- internal/scorer/scorer_test.go | 30 ++--- internal/server/server_test.go | 14 ++- issues/export-by-severity.md | 85 ++++++++++++++ website-prompt.md | 18 +++ 26 files changed, 508 insertions(+), 225 deletions(-) create mode 100644 internal/pipeline/checks.go delete mode 100644 internal/pipeline/pool.go delete mode 100644 internal/pipeline/pool_test.go create mode 100644 issues/export-by-severity.md create mode 100644 website-prompt.md diff --git a/README.md b/README.md index a855fb5..1bde8f0 100644 --- a/README.md +++ b/README.md @@ -264,10 +264,10 @@ A manifest holds the questions the model is asked: - `questions` is sent to the model unchanged, all questions, in the manifest's order, in one call. The reason question's chosen option is the `reason`. -- `decision`, optional, says how the yes/no (`noul`) answers make `p_keep` - (see [Combining the answers](#combining-the-answers)). Without it, `p_keep` - is the decision question's answer. It stays in the engine and is not sent - to the model. +- `decision`, optional, lists more yes/no (`noul`) questions that can keep a + trace (see [Any question can keep a trace](#any-question-can-keep-a-trace)). + Without it, the decision question decides alone. It stays in the engine + and is not sent to the model. - A relative `scorer.manifest` path resolves against the config file's directory. @@ -276,54 +276,68 @@ is asked the built-in questions ([`builtin_manifest.json`](internal/config/builtin_manifest.json)), and the engine logs a warning at start-up. They are: -| Question | Weight | Asks | -| --- | --- | --- | -| `answer_wrong` | 0.5 | Is the final answer wrong or only partially correct? | -| `unsupported_answer` | 0.1 | Does it make claims the evidence does not support? | -| `missing_tool_calls` | 0.1 | Did the agent skip lookups the task needed? | -| `answered_on_failure` | 0.1 | Did it answer despite a failed or partial tool call? | -| `ignored_instructions` | 0.1 | Does it go against the user's instructions? | -| `contradicts_knowledge` | 0.1 | Does it contradict general knowledge? | -| `why_wrong` | | Which [reason](#reasons) explains what is wrong? | +| Question | Asks | +| --- | --- | +| `answer_wrong` | Is the final answer wrong or only partially correct? | +| `unsupported_answer` | Does it make claims the evidence does not support? | +| `missing_tool_calls` | Did the agent skip lookups the task needed? | +| `answered_on_failure` | Did it answer despite a failed or partial tool call? | +| `ignored_instructions` | Does it go against the user's instructions? | +| `contradicts_knowledge` | Does it contradict general knowledge? | +| `why_wrong` | Which [reason](#reasons) explains what is wrong? | Each yes/no answer is the probability that the agent failed in that way, and -the weights combine them into `p_keep`. To word the questions +any of them can keep the trace. To word the questions for your agent, start from [`examples/manifests/my-agent.json`](examples/manifests/my-agent.json) and set `scorer.manifest` (or `TAILWISE_MANIFEST`). -### Combining the answers +### Any question can keep a trace -A manifest's `decision` block weights its yes/no questions: +A manifest's `decision` block lists the yes/no questions that can keep a +trace besides the decision question, in priority order, and can give any of +them its own threshold: ```json "decision": { - "intercept": 0, - "weights": {"answer_wrong": 0.5, "unsupported_answer": 0.1, "missing_tool_calls": 0.1, - "answered_on_failure": 0.1, "ignored_instructions": 0.1, "contradicts_knowledge": 0.1} + "any_of": ["unsupported_answer", "missing_tool_calls", "answered_on_failure", + "ignored_instructions", "contradicts_knowledge"], + "thresholds": {"contradicts_knowledge": 0.7} } ``` -and `p_keep` is a weighted [logarithmic opinion pool](https://en.wikipedia.org/wiki/Logarithmic_opinion_pool) -of their answers: - -``` -p_keep = sigmoid(intercept + Σ weight × logit(answer)) logit(p) = ln(p / (1 − p)) -``` - -- **Weights summing to 1, no intercept** (as in the built-in questions): - `p_keep` stays on the answers' own scale. Answers that agree keep their - probability (all at 0.9 give 0.9); answers that disagree meet in between. - A larger weight gives a question more say. -- **Fitted weights.** The formula is a logistic regression on the answers' - logits, so the coefficients and intercept of one fitted on labelled traces - can be used as they are. They need not sum to 1. -- **Rules.** Every weighted question must be a `noul` question of the same - manifest, and the model must answer all of them: a missing answer is a - scoring error, like a missing decision answer. Questions without a weight - are still asked and logged, but do not change `p_keep`. -- **Changing the weights changes `p_keep`,** so recheck `policy.keep_threshold` - against the share of traces kept. +- **Any question keeps.** A trace is kept when the decision question, or any + question in `any_of`, reaches its threshold. Each question checks a + different way an agent fails, and a wrong answer usually fails in one way + only: a check that says "no" is not evidence the answer is right, so it + never counts against keeping. +- **Thresholds.** `thresholds` sets a question's own threshold, in (0, 1]; + a question without one uses `policy.keep_threshold`. Raise it for a check + that flags too often. The decision question can have one too. +- **Flags.** The questions that reached their threshold are the trace's + flags: the problems the model found, in priority order. They are in the + decision log line (`flags`) and on the exported trace (`eval.flags`), next + to every answer (`eval.answers.`). A reviewer can filter on them, + for example every kept trace flagged `answered_on_failure`. +- **Priority and severity.** The decision question always ranks first, then + `any_of` in the order written. Ranks become weights that sum to 1, falling + steeply from the top ([rank-order centroid](https://doi.org/10.1287/mnsc.42.11.1515) + weights: with all six built-in questions, 0.41, 0.24, 0.16, 0.10, 0.06 and + 0.03). A trace's **severity** is the sum of weight × answer over its flags: + higher for more problems, and for higher-priority ones. It does not change + whether a trace is kept; it ranks kept traces for review (`severity` in the + log, `eval.severity` on the export). +- `p_keep` is the highest answer among the deciding questions. +- Every listed question must be a `noul` question of the same manifest, + listed once, and the model must answer all of them: a missing answer is a + scoring error, like a missing decision answer. Questions not listed are + still asked, logged and exported, but do not keep a trace. +- Each listed question is one more way to be kept, so listing more keeps more + traces. Recheck the share kept after changing the list, a threshold or a + question's wording. + +Without a `decision` block, the decision question decides alone, against +`policy.keep_threshold`. ### One model per agent @@ -617,16 +631,18 @@ so it does not take the engine out of a load balancer; watch queue for redelivery. On the synchronous endpoint, which has no queue, the engine retries it with `scorer.max_retries` and jittered `scorer.backoff`. The two kinds of retry never multiply. -4. **Decide.** `p_keep` combines the model's yes/no answers with the - manifest's weights (see [Combining the answers](#combining-the-answers)), - or, for a manifest without them, is its answer to the decision question. - - `p_keep >= keep_threshold`: keep, rule `above_threshold`. +4. **Decide.** The trace is kept when any question that decides (the + decision question, and those in the manifest's `decision.any_of`) + reaches its threshold; those questions are its flags (see + [Any question can keep a trace](#any-question-can-keep-a-trace)). + - A question reached its threshold: keep, rule `above_threshold`. - Otherwise keep with probability `random_floor`, rule `random_floor`. - Otherwise drop, rule `below_threshold`. 5. **Export.** Kept traces are queued to every exporter. Each decision produces one JSON log line with `trace_id`, `agent`, `decision`, `rule`, `reason`, `cause`, `p_keep`, `model`, `latency_ms`, - and `answers`: every yes/no answer, by question, as in - `"answers": {"answer_wrong": 0.91, "unsupported_answer": 0.84, ...}`. + `answers`, every yes/no answer by question, `flags`, the deciding + questions that reached their threshold, and `severity`, as in + `"answers": {"answer_wrong": 0.91, "unsupported_answer": 0.84, ...}, "flags": ["answer_wrong", "unsupported_answer"], "severity": 0.57`. ### Reasons @@ -899,7 +915,10 @@ Root span attributes: | `eval.cause` | string | both | The model's short name for the reason, e.g. `contradicts_evidence`; omitted when it gave none | | `eval.model` | string | both | The scorer model that decided | | `eval.latency_ms` | double | both | Scoring time in milliseconds | -| `eval.p_keep` | double | both | The model's probability that the answer is wrong; omitted on a fallback keep | +| `eval.p_keep` | double | both | The highest answer among the questions that decide; omitted on a fallback keep | +| `eval.flags` | string | both | The deciding questions that reached their threshold, comma-separated in priority order, e.g. `answer_wrong,unsupported_answer`; omitted when none did | +| `eval.severity` | double | both | Sum of priority weight × answer over the flags; omitted when there are none | +| `eval.answers.` | double | both | Each yes/no answer, e.g. `eval.answers.missing_tool_calls`; omitted on a fallback keep | Tool span attributes (rebuilt traces only; forwarded tool spans are not changed): @@ -1004,11 +1023,13 @@ These are open problems in the current version, not design choices to rely on. a change to its request schema), every trace is decided at random, about 98% dropped with the default `random_floor`, and only `WARN` log lines show it. -- **The built-in weights are a reasonable guess, not fitted.** Nothing tunes - them, or the model's probabilities, to your agent, so `keep_threshold` may - keep too much or too little at first. Watch the share of traces kept and - adjust the threshold; with labelled traces, fit the weights (see - [Combining the answers](#combining-the-answers)). +- **Decisions use the model's probabilities as they come.** Nothing tunes + them to your agent, and any of six questions can keep a trace, so the + defaults may keep too much at first. Watch the share of traces kept, and + which flags keep them, and adjust `decision.thresholds` or the `any_of` + list. +- **Severity ranks traces but nothing acts on it yet.** Exporters send kept + traces in the order they are decided, not by severity. - **Every scored trace is a billed call to Jev, and leaves your network.** Nothing caps the spend or redacts fields before sending; only `server.workers` limits the rate. diff --git a/architecture/sequence-diagrams.md b/architecture/sequence-diagrams.md index 9519baa..e5eb772 100644 --- a/architecture/sequence-diagrams.md +++ b/architecture/sequence-diagrams.md @@ -152,7 +152,7 @@ sequenceDiagram EW->>EW: decodeRecord(msg.Data()) → pipeline.Record EW->>T: Send(ctx, pipeline.Record) T->>B: Send(ctx, pipeline.Record) - Note over B: evalAttrs(Record) → eval.decision, rule, reason,
cause, model, p_keep, latency + Note over B: evalAttrs(Record) → eval.decision, rule, reason,
cause, model, p_keep, latency,
flags, answers.* alt Record.Trace.OTLP set (came in as spans) B->>O: Annotate(OTLP bytes, Annotation) else from the JSON API diff --git a/examples/manifests/my-agent.json b/examples/manifests/my-agent.json index b1aef32..e3da452 100644 --- a/examples/manifests/my-agent.json +++ b/examples/manifests/my-agent.json @@ -1,6 +1,6 @@ { "model": "jev-latest", - "note": "A manifest for your agent: the questions it is asked, all in one call. answer_wrong and the five checks after it are combined into p_keep with the weights in decision, and why_wrong picks the plain-English reason; word them all for your agent, and weight what matters to it.", + "note": "A manifest for your agent: the questions it is asked, all in one call. A trace is kept when answer_wrong, or any question in decision.any_of, reaches its threshold (decision.thresholds, else policy.keep_threshold); any_of is in priority order after answer_wrong, which ranks how severe a trace's problems are. why_wrong picks the plain-English reason. Word them all for your agent.", "questions": { "answer_wrong": { "type": "noul", @@ -41,14 +41,15 @@ } }, "decision": { - "intercept": 0, - "weights": { - "answer_wrong": 0.5, - "unsupported_answer": 0.1, - "missing_tool_calls": 0.1, - "answered_on_failure": 0.1, - "ignored_instructions": 0.1, - "contradicts_knowledge": 0.1 + "any_of": [ + "unsupported_answer", + "missing_tool_calls", + "answered_on_failure", + "ignored_instructions", + "contradicts_knowledge" + ], + "thresholds": { + "contradicts_knowledge": 0.7 } } } diff --git a/internal/config/builtin_manifest.json b/internal/config/builtin_manifest.json index fc08878..9b4bed6 100644 --- a/internal/config/builtin_manifest.json +++ b/internal/config/builtin_manifest.json @@ -1,6 +1,6 @@ { "model": "built-in", - "note": "The questions the engine asks when a model has no manifest of its own, all in one call: answer_wrong and the five checks after it are combined into p_keep with the weights in decision, and why_wrong picks the plain-English reason.", + "note": "The questions the engine asks when a model has no manifest of its own, all in one call. A trace is kept when answer_wrong, or any question in decision.any_of, reaches the keep threshold; any_of is in priority order after answer_wrong. why_wrong picks the plain-English reason.", "questions": { "answer_wrong": { "type": "noul", @@ -41,14 +41,12 @@ } }, "decision": { - "intercept": 0, - "weights": { - "answer_wrong": 0.5, - "unsupported_answer": 0.1, - "missing_tool_calls": 0.1, - "answered_on_failure": 0.1, - "ignored_instructions": 0.1, - "contradicts_knowledge": 0.1 - } + "any_of": [ + "unsupported_answer", + "missing_tool_calls", + "answered_on_failure", + "ignored_instructions", + "contradicts_knowledge" + ] } } diff --git a/internal/config/config.go b/internal/config/config.go index 2a64402..4569ef2 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -181,17 +181,21 @@ type Manifest struct { Model string `json:"model"` // Questions is kept as written (compacted), so keys reach the model in order. Questions json.RawMessage `json:"questions"` - // Decision, when set, combines the yes/no answers into p_keep; it stays in - // the engine and is not sent to the model. + // Decision, when set, lists more questions that decide; it stays in the + // engine and is not sent to the model. Decision *ManifestDecision `json:"decision"` Reasons map[string]string `json:"-"` } -// ManifestDecision weights each yes/no question's answer; see pipeline.Pool. +// ManifestDecision lists the yes/no questions that decide besides the +// decision question, in priority order after it: a trace is kept when any of +// their answers reaches its threshold. type ManifestDecision struct { - Intercept float64 `json:"intercept"` - Weights map[string]float64 `json:"weights"` + AnyOf []string `json:"any_of"` + // Thresholds overrides policy.keep_threshold for some of the deciding + // questions, the decision question included. + Thresholds map[string]float64 `json:"thresholds"` } // DefaultModel is what an agent's model starts from: the built-in scorer diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 9c40a18..c94765f 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "github.com/tecnosam/tailwise/internal/pipeline" "github.com/tecnosam/tailwise/internal/scorer" ) @@ -277,9 +278,9 @@ func TestLoadWithoutAFile(t *testing.T) { if m.Client.ReasonQuestion != "why_wrong" || m.Client.Reasons["contradicts_evidence"] != "The final answer contradicts the evidence the agent gathered." { t.Errorf("reason question %q, reasons %v", m.Client.ReasonQuestion, m.Client.Reasons) } - // They combine answer_wrong and the five checks into p_keep. - if p := m.Client.Pool; p == nil || len(p.Weights) != 6 || p.Weights["answer_wrong"] != 0.5 || p.Weights["contradicts_knowledge"] != 0.1 { - t.Errorf("pool = %+v, want the built-in weights", p) + // Any of the five checks can keep a trace, as well as answer_wrong. + if a := m.Client.AnyOf; len(a) != 5 || a[0] != "unsupported_answer" || a[4] != "contradicts_knowledge" { + t.Errorf("any_of = %q, want the five built-in checks", a) } t.Setenv(DefaultJevAPIKeyEnv, "") if _, err := Load(""); err == nil || !strings.Contains(err.Error(), "$JEV_API_KEY is not set") { @@ -385,36 +386,71 @@ func TestNATSNamesFromThePrefix(t *testing.T) { } } -// A manifest's decision block becomes the scorer's pool; it may weight only -// its own yes/no questions. Without one, p_keep is the decision answer. -func TestManifestDecisionWeights(t *testing.T) { +// A manifest's decision block lists more questions that decide; it may list +// only its own yes/no questions. +func TestManifestDecisionAnyOf(t *testing.T) { cfg, err := Load(write(t, "scorer:\n manifest: manifest.json\n", testManifest)) - if err != nil || cfg.Models.Default.Client.Pool != nil { - t.Fatalf("no decision block: pool %+v, error %v", cfg.Models.Default.Client.Pool, err) + if err != nil || cfg.Models.Default.Client.AnyOf != nil { + t.Fatalf("no decision block: any_of %q, error %v", cfg.Models.Default.Client.AnyOf, err) } with := func(decision string) string { return `{"questions": {"answer_wrong": {"type": "noul"}, "checks": {"type": "noul"},` + ` "quality": {"type": "score", "criteria": ["a", "b"]}}, "decision": ` + decision + `}` } - cfg, err = Load(write(t, "scorer:\n manifest: manifest.json\n", with(`{"intercept": -1, "weights": {"answer_wrong": 2, "checks": 0.5}}`))) + cfg, err = Load(write(t, "scorer:\n manifest: manifest.json\n", with(`{"any_of": ["checks"]}`))) if err != nil { t.Fatal(err) } - if p := cfg.Models.Default.Client.Pool; p == nil || p.Intercept != -1 || p.Weights["answer_wrong"] != 2 || p.Weights["checks"] != 0.5 { - t.Errorf("pool = %+v", p) + if a := cfg.Models.Default.Client.AnyOf; len(a) != 1 || a[0] != "checks" { + t.Errorf("any_of = %q", a) } - if strings.Contains(string(cfg.Models.Default.Client.Questions), "weights") { + if strings.Contains(string(cfg.Models.Default.Client.Questions), "any_of") { t.Errorf("the decision block reached the questions sent to the model: %s", cfg.Models.Default.Client.Questions) } for decision, wantErr := range map[string]string{ - `{"weights": {}}`: "must weight at least one question", - `{"weights": {"tone": 1}}`: `question "tone" not found`, - `{"weights": {"quality": 1}}`: `question "quality" has type "score", want "noul"`, + `{"any_of": []}`: "must list questions in any_of or set thresholds", + `{"any_of": ["checks", "checks"]}`: `"checks" is listed twice`, + `{"any_of": ["answer_wrong"]}`: `"answer_wrong" is listed twice, or is the decision question`, + `{"thresholds": {"checks": 0.6}}`: `"checks" is neither the decision question nor in any_of`, + `{"thresholds": {"answer_wrong": 0}}`: "want a number in (0, 1]", + `{"thresholds": {"answer_wrong": 1.5}}`: "want a number in (0, 1]", + `{"any_of": ["tone"]}`: `question "tone" not found`, + `{"any_of": ["quality"]}`: `question "quality" has type "score", want "noul"`, } { if _, err := Load(write(t, "scorer:\n manifest: manifest.json\n", with(decision))); err == nil || !strings.Contains(err.Error(), wantErr) { t.Errorf("%s: error = %v, want %q", decision, err, wantErr) } } } + +// The deciding questions become checks in priority order, the decision +// question first, weighted by rank, each with its own threshold or none. +func TestManifestDecisionChecks(t *testing.T) { + manifest := `{"questions": {"answer_wrong": {"type": "noul"}, "a": {"type": "noul"}, "b": {"type": "noul"}},` + + ` "decision": {"any_of": ["b", "a"], "thresholds": {"answer_wrong": 0.8, "a": 0.6}}}` + cfg, err := Load(write(t, "scorer:\n manifest: manifest.json\n", manifest)) + if err != nil { + t.Fatal(err) + } + checks := cfg.Models.Default.Checks + if len(checks) != 3 || checks[0].Question != "answer_wrong" || checks[1].Question != "b" || checks[2].Question != "a" { + t.Fatalf("checks = %+v, want answer_wrong, then any_of in order", checks) + } + if checks[0].Threshold != 0.8 || checks[1].Threshold != 0 || checks[2].Threshold != 0.6 { + t.Errorf("thresholds = %v %v %v, want 0.8, the policy's (0) and 0.6", checks[0].Threshold, checks[1].Threshold, checks[2].Threshold) + } + if !(checks[0].Weight > checks[1].Weight && checks[1].Weight > checks[2].Weight) { + t.Errorf("weights = %v %v %v, want them falling with priority", checks[0].Weight, checks[1].Weight, checks[2].Weight) + } + + // Without a decision block, the decision question is the one check. + cfg, err = Load(write(t, "scorer:\n manifest: manifest.json\n", testManifest)) + if err != nil { + t.Fatal(err) + } + if c := cfg.Models.Default.Checks; len(c) != 1 || c[0] != (pipeline.Check{Question: "answer_wrong", Weight: 1}) { + t.Errorf("checks = %+v, want answer_wrong alone with weight 1", c) + } +} diff --git a/internal/config/engine.go b/internal/config/engine.go index 6af738c..848efd0 100644 --- a/internal/config/engine.go +++ b/internal/config/engine.go @@ -140,13 +140,14 @@ func model(s Scorer, apiKey string, m *Manifest) scorer.ModelConfig { Model: s.Model, Questions: m.Questions, DecisionQuestion: s.DecisionQuestion, - Pool: m.pool(), + AnyOf: m.anyOf(), ReasonQuestion: reasonQuestion(s, m), Reasons: m.Reasons, APIKey: apiKey, Timeout: s.Timeout, }, - Retry: retry.Policy{Attempts: s.MaxRetries + 1, Backoff: s.Backoff, Jitter: true}, + Retry: retry.Policy{Attempts: s.MaxRetries + 1, Backoff: s.Backoff, Jitter: true}, + Checks: m.checks(s.DecisionQuestion), } } diff --git a/internal/config/manifest.go b/internal/config/manifest.go index 37267c0..ef6efb9 100644 --- a/internal/config/manifest.go +++ b/internal/config/manifest.go @@ -6,8 +6,8 @@ import ( _ "embed" "encoding/json" "fmt" - "maps" "os" + "slices" "github.com/tecnosam/tailwise/internal/pipeline" ) @@ -102,37 +102,65 @@ func (m *Manifest) prepare(decisionQuestion string) error { if head.Type != questionTypeNoul { return fmt.Errorf("decision question %q has type %q, want %q", decisionQuestion, head.Type, questionTypeNoul) } - return m.checkDecision(questions) + return m.checkDecision(questions, decisionQuestion) } -// checkDecision accepts a decision block whose weights name only yes/no -// questions of the manifest. -func (m *Manifest) checkDecision(questions map[string]json.RawMessage) error { +// checkDecision accepts a decision block that lists only yes/no questions of +// the manifest, each once, and sets thresholds only for questions that decide. +func (m *Manifest) checkDecision(questions map[string]json.RawMessage, decisionQuestion string) error { if m.Decision == nil { return nil } - if len(m.Decision.Weights) == 0 { - return fmt.Errorf("\"decision.weights\" must weight at least one question") + if len(m.Decision.AnyOf) == 0 && len(m.Decision.Thresholds) == 0 { + return fmt.Errorf("\"decision\" must list questions in any_of or set thresholds") } - for q := range m.Decision.Weights { + deciding := map[string]bool{decisionQuestion: true} + for _, q := range m.Decision.AnyOf { + if deciding[q] { + return fmt.Errorf("decision.any_of: question %q is listed twice, or is the decision question, which always comes first", q) + } + deciding[q] = true raw, ok := questions[q] if !ok { - return fmt.Errorf("decision.weights: question %q not found in questions", q) + return fmt.Errorf("decision.any_of: question %q not found in questions", q) } var head struct { Type string `json:"type"` } if err := json.Unmarshal(raw, &head); err != nil || head.Type != questionTypeNoul { - return fmt.Errorf("decision.weights: question %q has type %q, want %q", q, head.Type, questionTypeNoul) + return fmt.Errorf("decision.any_of: question %q has type %q, want %q", q, head.Type, questionTypeNoul) + } + } + for q, t := range m.Decision.Thresholds { + if !deciding[q] { + return fmt.Errorf("decision.thresholds: question %q is neither the decision question nor in any_of", q) + } + if t <= 0 || t > 1 { + return fmt.Errorf("decision.thresholds: %q is %v, want a number in (0, 1]", q, t) } } return nil } -// pool is the decision block as the scorer uses it, nil without one. -func (m *Manifest) pool() *pipeline.Pool { +// anyOf is the decision block's questions, none without one. +func (m *Manifest) anyOf() []string { if m.Decision == nil { return nil } - return &pipeline.Pool{Intercept: m.Decision.Intercept, Weights: maps.Clone(m.Decision.Weights)} + return slices.Clone(m.Decision.AnyOf) +} + +// checks are the questions that decide, the decision question first and then +// any_of in order, weighted by that priority. +func (m *Manifest) checks(decisionQuestion string) []pipeline.Check { + questions := append([]string{decisionQuestion}, m.anyOf()...) + weights := pipeline.PriorityWeights(len(questions)) + checks := make([]pipeline.Check, len(questions)) + for i, q := range questions { + checks[i] = pipeline.Check{Question: q, Weight: weights[i]} + if m.Decision != nil { + checks[i].Threshold = pipeline.Probability(m.Decision.Thresholds[q]) + } + } + return checks } diff --git a/internal/export/natsq/queue.go b/internal/export/natsq/queue.go index c5869a1..271c127 100644 --- a/internal/export/natsq/queue.go +++ b/internal/export/natsq/queue.go @@ -123,8 +123,12 @@ type recordMessage struct { Rule string `json:"rule"` Reason string `json:"reason"` Cause string `json:"cause,omitempty"` + Severity float64 `json:"severity,omitempty"` Model string `json:"model"` LatencyNS int64 `json:"latency_ns"` + + Answers map[string]pipeline.Probability `json:"answers,omitempty"` + Flags []string `json:"flags,omitempty"` } // A trace too big for the server keeps its decision but loses its original @@ -133,7 +137,8 @@ func encodeRecord(rec pipeline.Record, maxPayload int) ([]byte, error) { d := rec.Decision m := recordMessage{ Trace: rec.Trace, Agent: string(rec.Agent), Keep: d.Keep, Scored: d.Scored, PKeep: float64(d.PKeep), - Rule: string(d.Rule), Reason: d.Reason, Cause: d.Cause, Model: string(rec.Model), LatencyNS: int64(rec.Latency), + Rule: string(d.Rule), Reason: d.Reason, Cause: d.Cause, Severity: d.Severity, Model: string(rec.Model), LatencyNS: int64(rec.Latency), + Answers: rec.Answers, Flags: rec.Flags, } data, err := json.Marshal(m) if err != nil || maxPayload <= 0 || len(data) <= maxPayload || len(m.Trace.OTLP) == 0 { @@ -153,9 +158,11 @@ func decodeRecord(data []byte) (pipeline.Record, error) { Agent: pipeline.AgentLabel(m.Agent), Decision: pipeline.Decision{ Keep: m.Keep, Scored: m.Scored, PKeep: pipeline.Probability(m.PKeep), - Rule: pipeline.Rule(m.Rule), Reason: m.Reason, Cause: m.Cause, + Rule: pipeline.Rule(m.Rule), Reason: m.Reason, Cause: m.Cause, Severity: m.Severity, }, Model: pipeline.ModelName(m.Model), Latency: time.Duration(m.LatencyNS), + Answers: m.Answers, + Flags: m.Flags, }, nil } diff --git a/internal/export/natsq/queue_test.go b/internal/export/natsq/queue_test.go index 88fc694..bf5112d 100644 --- a/internal/export/natsq/queue_test.go +++ b/internal/export/natsq/queue_test.go @@ -3,6 +3,8 @@ package natsq import ( "context" "errors" + "maps" + "slices" "sync" "testing" "time" @@ -105,9 +107,11 @@ func TestRecordSurvivesTheStream(t *testing.T) { want := pipeline.Record{ Trace: trace.Trace{TraceID: "t1", Agent: "support-bot", Question: "q", FinalAnswer: "a", OTLP: []byte{1, 2, 3}}, Agent: "support-bot", - Decision: keepHigh, + Decision: pipeline.Decision{Keep: true, Scored: true, PKeep: 0.95, Rule: pipeline.RuleAboveThreshold, Severity: 0.57}, Model: "kev-latest", Latency: 93 * time.Millisecond, + Answers: map[string]pipeline.Probability{"answer_wrong": 0.95, "unsupported_answer": 0.75}, + Flags: []string{"answer_wrong", "unsupported_answer"}, } if err := q.Enqueue(context.Background(), want); err != nil { t.Fatal(err) @@ -115,7 +119,8 @@ func TestRecordSurvivesTheStream(t *testing.T) { select { case rec := <-got: if rec.Trace.TraceID != "t1" || rec.Trace.Agent != "support-bot" || rec.Agent != "support-bot" || string(rec.Trace.OTLP) != "\x01\x02\x03" || - rec.Decision != want.Decision || rec.Model != want.Model || rec.Latency != want.Latency { + rec.Decision != want.Decision || rec.Model != want.Model || rec.Latency != want.Latency || + !maps.Equal(rec.Answers, want.Answers) || !slices.Equal(rec.Flags, want.Flags) { t.Errorf("received %+v, want %+v", rec, want) } case <-time.After(10 * time.Second): diff --git a/internal/export/otlpbackend/backend.go b/internal/export/otlpbackend/backend.go index a7cd75a..c5930fb 100644 --- a/internal/export/otlpbackend/backend.go +++ b/internal/export/otlpbackend/backend.go @@ -7,6 +7,8 @@ import ( "crypto/rand" "fmt" "maps" + "slices" + "strings" "time" "go.opentelemetry.io/otel" @@ -108,6 +110,13 @@ func evalAttrs(rec pipeline.Record) []*commonpb.KeyValue { if d.Scored { attrs = append(attrs, otlp.FloatAttr("eval.p_keep", float64(d.PKeep))) } + if len(rec.Flags) > 0 { + attrs = append(attrs, otlp.StringAttr("eval.flags", strings.Join(rec.Flags, ",")), + otlp.FloatAttr("eval.severity", d.Severity)) + } + for _, q := range slices.Sorted(maps.Keys(rec.Answers)) { + attrs = append(attrs, otlp.FloatAttr("eval.answers."+q, float64(rec.Answers[q]))) + } return attrs } diff --git a/internal/export/otlpbackend/backend_test.go b/internal/export/otlpbackend/backend_test.go index b4a9c54..d3bb8d3 100644 --- a/internal/export/otlpbackend/backend_test.go +++ b/internal/export/otlpbackend/backend_test.go @@ -49,9 +49,11 @@ func TestExportKeptTrace(t *testing.T) { before := time.Now() err := exp.Send(context.Background(), pipeline.Record{ Trace: sampleTrace(), - Decision: keepHigh, + Decision: withSeverity(keepHigh, 0.8), Model: "kev-latest", Latency: 93 * time.Millisecond, + Answers: map[string]pipeline.Probability{"answer_wrong": 0.95, "unsupported_answer": 0.75}, + Flags: []string{"answer_wrong", "unsupported_answer"}, }) if err != nil { t.Fatal(err) @@ -87,6 +89,7 @@ func TestExportKeptTrace(t *testing.T) { "tailwise.evidence.1": "The Seine flows through Paris.", "eval.decision": "keep", "eval.rule": "above_threshold", + "eval.flags": "answer_wrong,unsupported_answer", "eval.model": "kev-latest", } @@ -95,7 +98,8 @@ func TestExportKeptTrace(t *testing.T) { t.Errorf("root %s = %q, want %q", k, got, want) } } - for k, want := range map[string]float64{"eval.p_keep": 0.95, "eval.latency_ms": 93} { + for k, want := range map[string]float64{"eval.p_keep": 0.95, "eval.latency_ms": 93, + "eval.answers.answer_wrong": 0.95, "eval.answers.unsupported_answer": 0.75, "eval.severity": 0.8} { if got := a[k].GetDoubleValue(); got != want { t.Errorf("root %s = %v, want %v", k, got, want) } @@ -246,3 +250,8 @@ func TestBackendReceivesOTLP(t *testing.T) { } } } + +func withSeverity(d pipeline.Decision, severity float64) pipeline.Decision { + d.Severity = severity + return d +} diff --git a/internal/pipeline/checks.go b/internal/pipeline/checks.go new file mode 100644 index 0000000..7da4388 --- /dev/null +++ b/internal/pipeline/checks.go @@ -0,0 +1,25 @@ +package pipeline + +// Check is one question that can keep a trace: an answer at or above its +// threshold flags the trace with the question. +type Check struct { + Question string + // Threshold is 0 to use the policy's keep threshold. + Threshold Probability + // Weight is the question's share of severity, from its priority; see + // PriorityWeights. + Weight float64 +} + +// PriorityWeights gives n questions, ranked 1 (most important) to n, their +// rank-order centroid weights: wᵢ = (1/n) Σ_{k=i..n} 1/k. They sum to 1 and +// fall with rank, steeply at the top: for 3 questions 0.61, 0.28 and 0.11. +func PriorityWeights(n int) []float64 { + w := make([]float64, n) + sum := 0.0 + for i := n; i >= 1; i-- { + sum += 1 / float64(i) + w[i-1] = sum / float64(n) + } + return w +} diff --git a/internal/pipeline/engine.go b/internal/pipeline/engine.go index 447f2d0..2e7447b 100644 --- a/internal/pipeline/engine.go +++ b/internal/pipeline/engine.go @@ -105,12 +105,13 @@ func (e *Engine) score(ctx context.Context, t trace.Trace, model Model, queued b } func (e *Engine) scored(t trace.Trace, model Model, score Score) Record { - d := e.deps.Policy.Decide(score.P) + d, flags := e.deps.Policy.DecideChecks(score.P, model.Checks, score.Deciding) d.Cause, d.Reason = score.Cause, score.Explanation if d.Reason == "" { d.Reason = probabilityReason(score.P) } - return Record{Trace: t, Agent: model.Agent, Decision: d, Model: score.Model, Latency: score.Latency, Answers: score.Answers} + return Record{Trace: t, Agent: model.Agent, Decision: d, Model: score.Model, Latency: score.Latency, + Answers: score.Answers, Flags: flags} } // probabilityReason is the reason for a model that gave none of its own. @@ -162,6 +163,9 @@ func (e *Engine) logDecision(out Outcome) { if d.Scored { attrs = append(attrs, "p_keep", float64(d.PKeep)) } + if rec.Flags != nil { + attrs = append(attrs, "flags", rec.Flags, "severity", d.Severity) + } if len(rec.Answers) > 0 { attrs = append(attrs, "answers", rec.Answers) } diff --git a/internal/pipeline/model.go b/internal/pipeline/model.go index 956bb00..c1e11bc 100644 --- a/internal/pipeline/model.go +++ b/internal/pipeline/model.go @@ -16,6 +16,9 @@ type Model struct { // Retry is used on the synchronous path only; the queued path leaves retrying // to the queue, so the two never multiply. Retry retry.Policy + // Checks are the questions that can keep a trace, in priority order. With + // none, p_keep alone decides. + Checks []Check // Agent is set by Route. Agent AgentLabel } diff --git a/internal/pipeline/policy.go b/internal/pipeline/policy.go index 5a7527c..21b34c1 100644 --- a/internal/pipeline/policy.go +++ b/internal/pipeline/policy.go @@ -24,13 +24,35 @@ func NewPolicy(cfg PolicyConfig, rnd func() float64) *Policy { func (p *Policy) Decide(pKeep Probability) Decision { d := Decision{Scored: true, PKeep: pKeep} - d.Keep, d.Rule = p.keep(pKeep) + d.Keep, d.Rule = p.keepIf(pKeep >= p.cfg.KeepThreshold) return d } -func (p *Policy) keep(pKeep Probability) (bool, Rule) { +// DecideChecks keeps a trace that any check flags. Flags are in the checks' +// order. With no checks, p_keep alone decides, as in Decide. +func (p *Policy) DecideChecks(pKeep Probability, checks []Check, answers map[string]Probability) (Decision, []string) { + if len(checks) == 0 { + return p.Decide(pKeep), nil + } + d := Decision{Scored: true, PKeep: pKeep} + flags := []string{} + for _, c := range checks { + threshold := c.Threshold + if threshold == 0 { + threshold = p.cfg.KeepThreshold + } + if a, ok := answers[c.Question]; ok && a >= threshold { + flags = append(flags, c.Question) + d.Severity += c.Weight * float64(a) + } + } + d.Keep, d.Rule = p.keepIf(len(flags) > 0) + return d, flags +} + +func (p *Policy) keepIf(above bool) (bool, Rule) { switch { - case pKeep >= p.cfg.KeepThreshold: + case above: return true, RuleAboveThreshold case p.sampled(): return true, RuleRandomFloor diff --git a/internal/pipeline/policy_test.go b/internal/pipeline/policy_test.go index e8d81f0..d739e93 100644 --- a/internal/pipeline/policy_test.go +++ b/internal/pipeline/policy_test.go @@ -1,6 +1,8 @@ package pipeline import ( + "math" + "slices" "testing" ) @@ -98,3 +100,66 @@ func TestDefaultRandomSourceRate(t *testing.T) { t.Errorf("random-floor keep rate = %.3f, want about 0.20", rate) } } + +func TestPriorityWeights(t *testing.T) { + for n, want := range map[int][]float64{ + 1: {1}, + 2: {0.75, 0.25}, + 3: {0.611, 0.278, 0.111}, + } { + got := PriorityWeights(n) + sum := 0.0 + for i := range got { + sum += got[i] + if math.Abs(got[i]-want[i]) > 1e-3 { + t.Errorf("PriorityWeights(%d) = %v, want %v", n, got, want) + break + } + } + if math.Abs(sum-1) > 1e-9 { + t.Errorf("PriorityWeights(%d) sums to %v, want 1", n, sum) + } + } +} + +// Any check reaching its own threshold keeps the trace and flags it; the +// flagged checks' weight × answer add up to its severity. +func TestDecideChecks(t *testing.T) { + checks := []Check{ + {Question: "answer_wrong", Weight: 0.6}, + {Question: "unsupported", Threshold: 0.8, Weight: 0.3}, + {Question: "missing_tools", Weight: 0.1}, + } + tests := []struct { + name string + answers map[string]Probability + draw []float64 + wantRule Rule + wantFlags []string + wantSeverity float64 + }{ + {"one low-priority check", map[string]Probability{"answer_wrong": 0.2, "unsupported": 0.1, "missing_tools": 0.9}, + nil, RuleAboveThreshold, []string{"missing_tools"}, 0.09}, + {"flags in priority order", map[string]Probability{"answer_wrong": 0.5, "unsupported": 0.9, "missing_tools": 1}, + nil, RuleAboveThreshold, []string{"answer_wrong", "unsupported", "missing_tools"}, 0.3 + 0.27 + 0.1}, + {"below its own threshold", map[string]Probability{"answer_wrong": 0.2, "unsupported": 0.7, "missing_tools": 0.1}, + []float64{0.9}, RuleBelowThreshold, []string{}, 0}, + {"random floor keeps with no flags", map[string]Probability{"answer_wrong": 0.2, "unsupported": 0.1, "missing_tools": 0.1}, + []float64{0.01}, RuleRandomFloor, []string{}, 0}, + } + for _, tt := range tests { + rnd, _ := fixed(t, tt.draw...) + d, flags := NewPolicy(testConfig, rnd).DecideChecks(0.9, checks, tt.answers) + if d.Rule != tt.wantRule || !slices.Equal(flags, tt.wantFlags) || math.Abs(d.Severity-tt.wantSeverity) > 1e-9 { + t.Errorf("%s: rule %s, flags %q, severity %v; want %s, %q, %v", tt.name, d.Rule, flags, d.Severity, tt.wantRule, tt.wantFlags, tt.wantSeverity) + } + if !d.Scored || d.PKeep != 0.9 { + t.Errorf("%s: p_keep not recorded: %+v", tt.name, d) + } + } + + // With no checks, p_keep alone decides. + if d, flags := NewPolicy(testConfig, nil).DecideChecks(0.6, nil, nil); !d.Keep || d.Rule != RuleAboveThreshold || flags != nil { + t.Errorf("no checks: %+v, flags %q", d, flags) + } +} diff --git a/internal/pipeline/pool.go b/internal/pipeline/pool.go deleted file mode 100644 index d307aff..0000000 --- a/internal/pipeline/pool.go +++ /dev/null @@ -1,40 +0,0 @@ -package pipeline - -import ( - "fmt" - "maps" - "math" - "slices" -) - -// Pool combines the model's answers into p_keep as a weighted logarithmic -// opinion pool: -// -// p_keep = sigmoid(Intercept + Σ Weights[q] · logit(answer q)) -// -// With weights summing to 1 and no intercept, answers that agree keep their -// probability and answers that disagree meet in between. A logistic -// regression fitted on the answers' logits has the same form, so its -// coefficients can be used as the weights. -type Pool struct { - Intercept float64 - Weights map[string]float64 -} - -// poolEpsilon keeps the logit of an answer of exactly 0 or 1 finite. -const poolEpsilon = 1e-6 - -// Combine needs an answer for every weighted question. Questions are summed -// in name order, so the same answers always give the same p_keep. -func (p *Pool) Combine(answers map[string]Probability) (Probability, error) { - z := p.Intercept - for _, q := range slices.Sorted(maps.Keys(p.Weights)) { - a, ok := answers[q] - if !ok { - return 0, fmt.Errorf("model response has no answer for %q", q) - } - x := min(max(float64(a), poolEpsilon), 1-poolEpsilon) - z += p.Weights[q] * math.Log(x/(1-x)) - } - return Probability(1 / (1 + math.Exp(-z))), nil -} diff --git a/internal/pipeline/pool_test.go b/internal/pipeline/pool_test.go deleted file mode 100644 index 3d0af57..0000000 --- a/internal/pipeline/pool_test.go +++ /dev/null @@ -1,36 +0,0 @@ -package pipeline - -import ( - "math" - "strings" - "testing" -) - -func TestPoolCombines(t *testing.T) { - half := &Pool{Weights: map[string]float64{"a": 0.5, "b": 0.5}} - for _, tt := range []struct { - name string - pool *Pool - answers map[string]Probability - want float64 - }{ - {"agreeing answers keep their probability", half, map[string]Probability{"a": 0.9, "b": 0.9}, 0.9}, - {"opposite answers meet in the middle", half, map[string]Probability{"a": 0.9, "b": 0.1}, 0.5}, - {"one question with weight 1 is its answer", &Pool{Weights: map[string]float64{"a": 1}}, map[string]Probability{"a": 0.3}, 0.3}, - {"the intercept shifts the log-odds", &Pool{Intercept: math.Log(9), Weights: map[string]float64{"a": 1}}, map[string]Probability{"a": 0.5}, 0.9}, - {"unweighted answers are ignored", half, map[string]Probability{"a": 0.2, "b": 0.2, "c": 1}, 0.2}, - {"an answer of exactly 1 stays finite", half, map[string]Probability{"a": 1, "b": 0.5}, 0.999}, - } { - got, err := tt.pool.Combine(tt.answers) - if err != nil || math.Abs(float64(got)-tt.want) > 1e-3 { - t.Errorf("%s: got %v, %v; want %v", tt.name, got, err, tt.want) - } - } -} - -func TestPoolNeedsEveryWeightedAnswer(t *testing.T) { - p := &Pool{Weights: map[string]float64{"a": 0.5, "b": 0.5}} - if _, err := p.Combine(map[string]Probability{"a": 0.9}); err == nil || !strings.Contains(err.Error(), `no answer for "b"`) { - t.Errorf("error = %v", err) - } -} diff --git a/internal/pipeline/types.go b/internal/pipeline/types.go index 77b88ea..d965aa7 100644 --- a/internal/pipeline/types.go +++ b/internal/pipeline/types.go @@ -24,8 +24,8 @@ const ( ) type Score struct { - // P is p_keep: the decision question's answer, or the manifest's pool of - // answers. + // P is p_keep: the highest answer among the questions that decide, the + // decision question alone unless the manifest lists more. P Probability Model ModelName Latency time.Duration // as the server reported it, else the round trip @@ -35,6 +35,9 @@ type Score struct { Explanation string // Answers holds every yes/no (noul) answer the model gave, by question. Answers map[string]Probability + // Deciding holds the answers of the questions that decide; P is their + // highest. + Deciding map[string]Probability } type Decision struct { @@ -49,6 +52,9 @@ type Decision struct { Reason string // Cause is bounded by the manifest, so safe as a metric attribute. Cause string + // Severity is the sum of the flagged checks' weight × answer: higher for + // more, and higher-priority, problems. 0 when nothing was flagged. + Severity float64 } type Record struct { @@ -57,9 +63,11 @@ type Record struct { Decision Decision Model ModelName Latency time.Duration - // Answers are the model's yes/no answers p_keep was decided from; none on - // fallback. + // Answers are the model's yes/no answers, none on fallback. Answers map[string]Probability + // Flags names the checks whose answer reached their threshold, in + // priority order: the problems the model found. + Flags []string } type Outcome struct { diff --git a/internal/scorer/factory.go b/internal/scorer/factory.go index 86448a6..d635552 100644 --- a/internal/scorer/factory.go +++ b/internal/scorer/factory.go @@ -12,6 +12,8 @@ type ModelConfig struct { Client Config // Retry is how the synchronous endpoint retries a failed call. Retry retry.Policy + // Checks are the questions that can keep a trace, in priority order. + Checks []pipeline.Check } type Models struct { @@ -38,5 +40,5 @@ func buildModel(mc ModelConfig) (pipeline.Model, error) { if err != nil { return pipeline.Model{}, err } - return pipeline.Model{Name: pipeline.ModelName(mc.Client.Model), Scorer: s, Retry: mc.Retry}, nil + return pipeline.Model{Name: pipeline.ModelName(mc.Client.Model), Scorer: s, Retry: mc.Retry, Checks: mc.Checks}, nil } diff --git a/internal/scorer/scorer.go b/internal/scorer/scorer.go index fb00108..a42971a 100644 --- a/internal/scorer/scorer.go +++ b/internal/scorer/scorer.go @@ -30,9 +30,9 @@ type Config struct { Model string Questions json.RawMessage DecisionQuestion string - // Pool, when set, combines the yes/no answers into p_keep; without it - // p_keep is the decision question's answer. - Pool *pipeline.Pool + // AnyOf lists the yes/no questions that decide: p_keep is their highest + // answer. Empty, the decision question decides alone. + AnyOf []string // Reasons maps each option of the reason question to the plain-English // sentence that becomes the decision's reason. ReasonQuestion string @@ -162,12 +162,16 @@ func (c *Client) parse(raw []byte, elapsed time.Duration) (pipeline.Score, error return pipeline.Score{}, err } answers := noulAnswers(out) - if c.cfg.Pool != nil { - if p, err = c.cfg.Pool.Combine(answers); err != nil { - return pipeline.Score{}, err + deciding := map[string]pipeline.Probability{c.cfg.DecisionQuestion: p} + for _, q := range c.cfg.AnyOf { + a, ok := answers[q] + if !ok { + return pipeline.Score{}, fmt.Errorf("model response has no answer for %q", q) } + deciding[q] = a + p = max(p, a) } - score := pipeline.Score{P: p, Answers: answers, Model: c.modelName(out), Latency: reportedLatency(out, elapsed)} + score := pipeline.Score{P: p, Answers: answers, Deciding: deciding, Model: c.modelName(out), Latency: reportedLatency(out, elapsed)} score.Cause, score.Explanation = c.reasonAnswer(out) return score, nil } diff --git a/internal/scorer/scorer_test.go b/internal/scorer/scorer_test.go index 47e5f25..15086c5 100644 --- a/internal/scorer/scorer_test.go +++ b/internal/scorer/scorer_test.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "io" - "math" "net/http" "net/http/httptest" "os" @@ -16,7 +15,6 @@ import ( "time" "github.com/tecnosam/tailwise/internal/fault" - "github.com/tecnosam/tailwise/internal/pipeline" ) const questions = `{"answer_wrong":{"type":"noul","instructions":"Is the final answer wrong?"},"answers_question":{"type":"score","instructions":"How good?","criteria":["Wrong","Fully correct"]}}` @@ -381,34 +379,30 @@ func TestScoreReadsTheReason(t *testing.T) { } } -// Every yes/no answer is kept; with a pool, p_keep combines them, and without -// one it is the decision question's answer. -func TestScoreKeepsAndPoolsTheAnswers(t *testing.T) { +// Every yes/no answer is kept. p_keep is the highest answer among the +// questions that decide: the decision question, and any listed in AnyOf. +func TestScoreKeepsTheAnswersAndTakesTheHighest(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Write([]byte(`{"answers": {"answer_wrong": {"type": "noul", "noul": 0.9}, "unsupported_answer": {"type": "noul", "noul": 0.1},` + + w.Write([]byte(`{"answers": {"answer_wrong": {"type": "noul", "noul": 0.2}, "unsupported_answer": {"type": "noul", "noul": 0.8},` + ` "answers_question": {"type": "score", "score": 1.2}, "bad": {"type": "noul", "noul": 3}}}`)) })) defer srv.Close() res, err := newClient(t, srv.URL, nil).Score(context.Background(), "s") - if err != nil || res.P != 0.9 { - t.Fatalf("without a pool: %+v, %v", res, err) + if err != nil || res.P != 0.2 || len(res.Deciding) != 1 { + t.Fatalf("decision question alone: %+v, %v", res, err) } - if len(res.Answers) != 2 || res.Answers["answer_wrong"] != 0.9 || res.Answers["unsupported_answer"] != 0.1 { + if len(res.Answers) != 2 || res.Answers["answer_wrong"] != 0.2 || res.Answers["unsupported_answer"] != 0.8 { t.Errorf("answers = %v, want the two well-formed yes/no answers", res.Answers) } - pooled := newClient(t, srv.URL, func(cfg *Config) { - cfg.Pool = &pipeline.Pool{Weights: map[string]float64{"answer_wrong": 0.5, "unsupported_answer": 0.5}} - }) - if res, err = pooled.Score(context.Background(), "s"); err != nil || math.Abs(float64(res.P)-0.5) > 1e-9 { - t.Errorf("with a pool: %+v, %v; want p_keep 0.5", res, err) + anyOf := newClient(t, srv.URL, func(cfg *Config) { cfg.AnyOf = []string{"unsupported_answer"} }) + if res, err = anyOf.Score(context.Background(), "s"); err != nil || res.P != 0.8 || len(res.Deciding) != 2 { + t.Errorf("with any_of: %+v, %v; want p_keep 0.8 from two deciding answers", res, err) } - missing := newClient(t, srv.URL, func(cfg *Config) { - cfg.Pool = &pipeline.Pool{Weights: map[string]float64{"answer_wrong": 0.5, "missing_tool_calls": 0.5}} - }) + missing := newClient(t, srv.URL, func(cfg *Config) { cfg.AnyOf = []string{"missing_tool_calls"} }) if _, err = missing.Score(context.Background(), "s"); err == nil || !strings.Contains(err.Error(), `no answer for "missing_tool_calls"`) { - t.Errorf("a weighted question without an answer: error = %v", err) + t.Errorf("a deciding question without an answer: error = %v", err) } } diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 023220b..a95bc66 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "log/slog" + "math" "net/http" "net/http/httptest" "os" @@ -657,10 +658,13 @@ func TestDecisionLogLine(t *testing.T) { var buf strings.Builder answered := func(context.Context, trace.State) (pipeline.Score, error) { return pipeline.Score{P: 0.9, Model: "kev-latest", Latency: 12 * time.Millisecond, - Answers: map[string]pipeline.Probability{"answer_wrong": 0.95, "unsupported_answer": 0.7}}, nil + Answers: map[string]pipeline.Probability{"answer_wrong": 0.95, "unsupported_answer": 0.7, "missing_tool_calls": 0.1}, + Deciding: map[string]pipeline.Probability{"answer_wrong": 0.95, "unsupported_answer": 0.7, "missing_tool_calls": 0.1}}, nil } + checks := []pipeline.Check{{Question: "answer_wrong", Weight: 0.6}, {Question: "unsupported_answer", Weight: 0.3}, + {Question: "missing_tool_calls", Weight: 0.1}} h := newHarness(t, harnessOpts{ - scorer: &fakeScorer{score: answered}, + models: pipeline.SingleModel(pipeline.Model{Name: "kev-latest", Scorer: &fakeScorer{score: answered}, Checks: checks}), logger: jsonLogger(&buf), }) h.do("POST", "/v1/traces:score", `{"trace_id": "log-1", "question": "q", "final_answer": "a"}`) @@ -683,6 +687,12 @@ func TestDecisionLogLine(t *testing.T) { if answers["answer_wrong"] != 0.95 || answers["unsupported_answer"] != 0.7 { t.Errorf("log field answers = %v, want every yes/no answer", line["answers"]) } + if flags, _ := json.Marshal(line["flags"]); string(flags) != `["answer_wrong","unsupported_answer"]` { + t.Errorf("log field flags = %s, want the deciding questions at or above the threshold", flags) + } + if s, _ := line["severity"].(float64); math.Abs(s-(0.6*0.95+0.3*0.7)) > 1e-9 { + t.Errorf("log field severity = %v, want the flagged checks' weight × answer", line["severity"]) + } } func deref(p *float64) any { diff --git a/issues/export-by-severity.md b/issues/export-by-severity.md new file mode 100644 index 0000000..642d8bf --- /dev/null +++ b/issues/export-by-severity.md @@ -0,0 +1,85 @@ +# Export kept traces in order of severity + +**Labels:** enhancement, export + +## Problem + +Kept traces are exported in the order they are decided. When more traces are +kept than reviewers can look at, the ones that reach them first should be the +ones with the worst problems, not the ones that happened to finish first. + +Take two kept traces: + +- Trace 1 is flagged `answer_wrong` and `unsupported_answer`. +- Trace 2 is flagged `answer_wrong` and `contradicts_knowledge`. + +Both are worth reviewing, but if `unsupported_answer` ranks above +`contradicts_knowledge` for this agent, trace 1 should go first. + +## What exists today + +- Each manifest's `decision` block ranks the questions that can keep a trace: + the decision question first, then `any_of` in the order written. Each + question can have its own threshold (`decision.thresholds`). +- Ranks become rank-order centroid weights (`pipeline.PriorityWeights`). +- Every kept trace gets a **severity**: the sum of weight × answer over its + flags (`pipeline.Policy.DecideChecks`). It is in the decision log + (`severity`) and on the exported root span (`eval.severity`). +- Nothing orders by severity yet. The export queues + (`internal/export/natsq`) are JetStream work queues and deliver first in, + first out. + +## Proposal + +Have each exporter send higher-severity traces first. + +JetStream streams are strictly ordered, so a priority queue needs one of these +designs: + +1. **Priority bands.** Publish each record to one of a few subjects by + severity (for example `exports..high`, `.medium` and `.low`), with + one consumer per band. Exporter workers pull from the highest band that + has messages. This is simple and durable, and the ordering inside a band + stays first in, first out. +2. **A reordering window.** Pull a batch, sort it by severity and send in + that order. This is exact within the window, and adds latency equal to + the window. +3. **Budgeted keeping.** Instead of keeping everything above threshold, keep + the top N by severity per period (for example per hour, per agent). This + matters most when review capacity is a fixed number rather than a share. + +Option 1 fits the current queue design best. + +## Things to decide + +- **Calibrating severity across questions.** Severity uses the raw answers. + A question with a high threshold (say 0.9) that just passes contributes + almost as much as one that passes a 0.5 threshold with a 0.95 answer. + Consider using how far each answer is past its threshold, + `(answer − threshold) / (1 − threshold)`, instead of the raw answer. Once + labelled traces exist, calibrate each question's answers against the + labels. +- **Starvation.** If high-severity traces keep arriving, low ones may never + be sent. Age records upward, or cap how long a record can wait. +- **Random-floor samples.** Traces kept by `random_floor` have no flags and a + severity of 0. They are the unbiased sample used to measure what the + checks miss, so they must not wait behind everything else forever. Give + them their own band, or a guaranteed share. +- **Is a queue needed at all?** If the back end can sort (Phoenix can filter + and sort on `eval.severity`), reviewers may get most of the value without + changing export. A priority queue matters when export itself is the + bottleneck, or when only part of the kept traces should be sent. + +## Related + +- Sampling per flag: a separate random floor for each question, so every + kind of problem gets its own sample for review. It would use the same + bands. + +## Done when + +- Exporters send higher-severity traces before lower ones, by one of the + designs above. +- Random-floor samples still reach every back end within a bounded delay. +- The README documents the ordering and how to tune it, and the tests cover + ordering, starvation and restarts (queued records keep their band). diff --git a/website-prompt.md b/website-prompt.md new file mode 100644 index 0000000..d501edb --- /dev/null +++ b/website-prompt.md @@ -0,0 +1,18 @@ + I want you to create a beautiful landing page for the tailwise project. call it tailwise-website. I want something minimalist, simple, but have a very immersive and smooth feel, with reasonable + amount of animations (thinking framer, three.js, etc). primary color is a nice shade of white. Secondary is a very unique and dense shade of purple mixed with pink that'll contrast nice on the + white. + + This should also have a dark mode. + + Stack, use tailwindCSS, NextJS and other stuff you need for those sexy animations and immersion i've been gooning to all over this prompt. + + Included: + + 1. what this project is + 2. Link to Github + 3. Link to releases + 4. Documentation + 5. other stuff a Devtool like this would need in its website. + + + /frontend-design \ No newline at end of file