feat(llmrec): 录制 LLM 请求/响应 HTTP 原文
归一化视图(request_body/response_body)会丢弃工具 schema、tool_use 块， 且看不到原始 SSE 帧，排查 provider 侧问题时信息不足。新增在 HTTP transport 层捕获真实 wire body 的能力： - llmrec: 新增 Capture，Recorder 挂到 ctx，quotaAwareTransport 填充。 请求取 GetBody 副本(不消费原 body)，响应用 tee 保持流式;norma 内部重试 的多次尝试逐次保留(旧代码会读掉丢弃),单次时为字节级原文。 - db: llm_records 增加 raw_request/raw_response 两列，含旧库 ALTER 迁移。 - web: 录制页详情面板增加「原文」视图切换，及请求/响应内容复制按钮 (兼容非安全上下文的 execCommand 回退)。 - 测试: 单元(capture)、端到端(逐字节比对捕获=服务器实际收到)、 迁移(真实 PG 事务回滚)、全链路(Recorder→provider→PG 落库)。
This commit is contained in:
Autumn-27 committed
1 parent
4d9580ebfa
commit
19a7206d12
12 files changed
+884
-17
No files matched your search
@@ -0,0 +1,140 @@
|
||||
package llmrec
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type captureContextKey struct{}
|
||||
|
||||
// Capture collects the untouched wire bodies of one logical LLM call. The
|
||||
// Recorder creates it and puts it on the context; the HTTP transport that norma
|
||||
// dials through (agent.quotaAwareTransport) finds it there and fills it in.
|
||||
//
|
||||
// This exists because everything the Recorder itself sees is already normalized:
|
||||
// llm.CompletionRequest is re-serialized rather than the body buildBody() sent,
|
||||
// and the response arrives as decoded StreamEvents, not the SSE frames. For
|
||||
// debugging a live provider, the bytes on the wire are the only ground truth.
|
||||
//
|
||||
// norma's doStream retries the request-establishment phase, so one Stream can
|
||||
// issue several HTTP attempts. Each attempt is kept: the discarded ones (see
|
||||
// norma/llm/retry.go, which closes non-final bodies unread) are exactly what
|
||||
// makes rate-limit and gateway failures diagnosable.
|
||||
//
|
||||
// The transport writes from norma's stream-reading goroutine while the Recorder
|
||||
// snapshots at stream end, so all state is mutex-guarded.
|
||||
type Capture struct {
|
||||
mu sync.Mutex
|
||||
request string
|
||||
attempts []*attempt
|
||||
}
|
||||
|
||||
type attempt struct {
|
||||
status int
|
||||
body strings.Builder
|
||||
}
|
||||
|
||||
// NewCapture returns a context carrying a fresh Capture, plus the Capture itself.
|
||||
func NewCapture(ctx context.Context) (context.Context, *Capture) {
|
||||
c := &Capture{}
|
||||
return context.WithValue(ctx, captureContextKey{}, c), c
|
||||
}
|
||||
|
||||
// CaptureFrom returns the Capture attached to ctx, or nil when raw capture is
|
||||
// off. Callers must tolerate nil — recording is a toggle, and non-recorded
|
||||
// providers dial through the same transport.
|
||||
func CaptureFrom(ctx context.Context) *Capture {
|
||||
if ctx == nil {
|
||||
return nil
|
||||
}
|
||||
c, _ := ctx.Value(captureContextKey{}).(*Capture)
|
||||
return c
|
||||
}
|
||||
|
||||
// SetRequest stores the outgoing request body. Retries re-send identical bytes,
|
||||
// so only the first attempt's body is kept.
|
||||
func (c *Capture) SetRequest(body string) {
|
||||
if c == nil || body == "" {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.request == "" {
|
||||
c.request = body
|
||||
}
|
||||
}
|
||||
|
||||
// TeeResponse opens a new attempt and wraps rc so everything read from it is
|
||||
// mirrored into that attempt. It tees rather than reads because a successful
|
||||
// response is an SSE stream that must keep streaming to the caller.
|
||||
func (c *Capture) TeeResponse(status int, rc io.ReadCloser) io.ReadCloser {
|
||||
if c == nil || rc == nil {
|
||||
return rc
|
||||
}
|
||||
a := &attempt{status: status}
|
||||
c.mu.Lock()
|
||||
c.attempts = append(c.attempts, a)
|
||||
c.mu.Unlock()
|
||||
return &teeBody{rc: rc, c: c, a: a}
|
||||
}
|
||||
|
||||
// RawRequest returns the request body as sent, or "" if nothing was captured.
|
||||
func (c *Capture) RawRequest() string {
|
||||
if c == nil {
|
||||
return ""
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.request
|
||||
}
|
||||
|
||||
// RawResponse returns the response bytes as received. A single attempt yields
|
||||
// the untouched original (copy-pasteable straight into a replay); multiple
|
||||
// attempts are concatenated behind per-attempt header lines so a retry sequence
|
||||
// stays readable.
|
||||
func (c *Capture) RawResponse() string {
|
||||
if c == nil {
|
||||
return ""
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
switch len(c.attempts) {
|
||||
case 0:
|
||||
return ""
|
||||
case 1:
|
||||
return c.attempts[0].body.String()
|
||||
}
|
||||
var b strings.Builder
|
||||
for i, a := range c.attempts {
|
||||
fmt.Fprintf(&b, "===== attempt %d/%d — HTTP %d =====\n", i+1, len(c.attempts), a.status)
|
||||
body := a.body.String()
|
||||
b.WriteString(body)
|
||||
if !strings.HasSuffix(body, "\n") {
|
||||
b.WriteByte('\n')
|
||||
}
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// teeBody mirrors reads into a Capture attempt, guarded by the Capture's mutex
|
||||
// so a snapshot taken mid-stream never races the writer.
|
||||
type teeBody struct {
|
||||
rc io.ReadCloser
|
||||
c *Capture
|
||||
a *attempt
|
||||
}
|
||||
|
||||
func (t *teeBody) Read(p []byte) (int, error) {
|
||||
n, err := t.rc.Read(p)
|
||||
if n > 0 {
|
||||
t.c.mu.Lock()
|
||||
t.a.body.Write(p[:n])
|
||||
t.c.mu.Unlock()
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (t *teeBody) Close() error { return t.rc.Close() }
|
||||
@@ -0,0 +1,103 @@
|
||||
package llmrec
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestCaptureFromNilAndMissing(t *testing.T) {
|
||||
var nilCtx context.Context // the transport reaches here on any unrecorded call
|
||||
if CaptureFrom(nilCtx) != nil {
|
||||
t.Fatal("nil context should yield no capture")
|
||||
}
|
||||
if CaptureFrom(context.Background()) != nil {
|
||||
t.Fatal("bare context should yield no capture")
|
||||
}
|
||||
}
|
||||
|
||||
// A nil *Capture is the "recording off" path: the transport calls these on every
|
||||
// request, so they must all be no-ops rather than panics.
|
||||
func TestNilCaptureIsInert(t *testing.T) {
|
||||
var c *Capture
|
||||
c.SetRequest("body")
|
||||
if got := c.RawRequest(); got != "" {
|
||||
t.Fatalf("RawRequest()=%q want empty", got)
|
||||
}
|
||||
if got := c.RawResponse(); got != "" {
|
||||
t.Fatalf("RawResponse()=%q want empty", got)
|
||||
}
|
||||
rc := io.NopCloser(strings.NewReader("x"))
|
||||
if c.TeeResponse(200, rc) != rc {
|
||||
t.Fatal("nil capture must pass the body through untouched")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCaptureSingleAttemptKeepsBytesVerbatim(t *testing.T) {
|
||||
ctx, c := NewCapture(context.Background())
|
||||
if CaptureFrom(ctx) != c {
|
||||
t.Fatal("capture not retrievable from its own context")
|
||||
}
|
||||
c.SetRequest(`{"model":"x"}`)
|
||||
// Retries re-send identical bytes; only the first is kept.
|
||||
c.SetRequest(`{"model":"ignored"}`)
|
||||
|
||||
const sse = "event: message_start\ndata: {\"type\":\"message_start\"}\n\nevent: message_stop\ndata: {}\n\n"
|
||||
body := c.TeeResponse(200, io.NopCloser(strings.NewReader(sse)))
|
||||
got, err := io.ReadAll(body)
|
||||
if err != nil {
|
||||
t.Fatalf("read: %v", err)
|
||||
}
|
||||
if string(got) != sse {
|
||||
t.Fatal("tee altered the stream delivered to the caller")
|
||||
}
|
||||
if err := body.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
|
||||
if c.RawRequest() != `{"model":"x"}` {
|
||||
t.Fatalf("RawRequest()=%q", c.RawRequest())
|
||||
}
|
||||
// A lone attempt must stay byte-identical — no headers, no framing added, so
|
||||
// it can be replayed as-is.
|
||||
if c.RawResponse() != sse {
|
||||
t.Fatalf("RawResponse()=%q want verbatim SSE", c.RawResponse())
|
||||
}
|
||||
}
|
||||
|
||||
// norma's doStream discards the bodies of retried attempts (retry.go closes them
|
||||
// unread), so the capture is the only place a 429 body survives.
|
||||
func TestCaptureRetriedAttemptsAreAllKept(t *testing.T) {
|
||||
_, c := NewCapture(context.Background())
|
||||
|
||||
first := c.TeeResponse(429, io.NopCloser(strings.NewReader(`{"error":"rate_limit"}`)))
|
||||
if _, err := io.ReadAll(first); err != nil {
|
||||
t.Fatalf("read first: %v", err)
|
||||
}
|
||||
second := c.TeeResponse(200, io.NopCloser(strings.NewReader("data: ok\n")))
|
||||
if _, err := io.ReadAll(second); err != nil {
|
||||
t.Fatalf("read second: %v", err)
|
||||
}
|
||||
|
||||
raw := c.RawResponse()
|
||||
if !strings.Contains(raw, `{"error":"rate_limit"}`) {
|
||||
t.Fatalf("dropped the retried attempt body: %q", raw)
|
||||
}
|
||||
if !strings.Contains(raw, "data: ok") {
|
||||
t.Fatalf("dropped the final attempt body: %q", raw)
|
||||
}
|
||||
if !strings.Contains(raw, "attempt 1/2 — HTTP 429") || !strings.Contains(raw, "attempt 2/2 — HTTP 200") {
|
||||
t.Fatalf("attempts not delimited: %q", raw)
|
||||
}
|
||||
if strings.Index(raw, "rate_limit") > strings.Index(raw, "data: ok") {
|
||||
t.Fatal("attempts stored out of order")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCaptureEmptyWhenNoHTTPHappened(t *testing.T) {
|
||||
_, c := NewCapture(context.Background())
|
||||
if c.RawRequest() != "" || c.RawResponse() != "" {
|
||||
t.Fatal("unused capture should be empty")
|
||||
}
|
||||
}
|
||||
+13
-2
@@ -110,9 +110,15 @@ func (r *Recorder) Stream(ctx context.Context, req llm.CompletionRequest) iter.S
|
||||
}
|
||||
|
||||
// Serialize the request only when storing bodies (this is the expensive part).
|
||||
// Alongside it, attach a Capture so the HTTP transport can hand back the
|
||||
// untouched wire bodies — the normalized view below cannot reconstruct them
|
||||
// (tool schemas are dropped, tool_use blocks never reach this layer, and the
|
||||
// SSE framing is already decoded). See capture.go.
|
||||
reqBody := ""
|
||||
var capt *Capture
|
||||
if recordBodies {
|
||||
reqBody = r.serializeRequest(req)
|
||||
ctx, capt = NewCapture(ctx)
|
||||
}
|
||||
|
||||
return func(yield func(llm.StreamEvent, error) bool) {
|
||||
@@ -133,7 +139,7 @@ func (r *Recorder) Stream(ctx context.Context, req llm.CompletionRequest) iter.S
|
||||
r.recordUsage(taskID, expID, worker, usage, int(time.Since(start).Milliseconds()), status)
|
||||
// Heavy trace row — only when body recording is on.
|
||||
if recordBodies {
|
||||
r.record(req, session, taskID, worker, reqBody, start, textBuf.String(), thinkingBuf.String(), usage, stopReason, err)
|
||||
r.record(req, session, taskID, worker, reqBody, capt, start, textBuf.String(), thinkingBuf.String(), usage, stopReason, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -210,7 +216,7 @@ func (r *Recorder) recordUsage(taskID string, expID int64, worker string, usage
|
||||
// record persists one LLM call to PostgreSQL before the provider stream returns.
|
||||
// Keeping the write inside the owning task operation means task deletion can
|
||||
// drain calls and then remove records without a late async insert recreating one.
|
||||
func (r *Recorder) record(req llm.CompletionRequest, session, taskID, worker, reqBody string, start time.Time, text, thinking string, usage llm.Usage, stopReason string, streamErr error) {
|
||||
func (r *Recorder) record(req llm.CompletionRequest, session, taskID, worker, reqBody string, capt *Capture, start time.Time, text, thinking string, usage llm.Usage, stopReason string, streamErr error) {
|
||||
latency := int(time.Since(start).Milliseconds())
|
||||
status := "ok"
|
||||
errMsg := ""
|
||||
@@ -256,6 +262,11 @@ func (r *Recorder) record(req llm.CompletionRequest, session, taskID, worker, re
|
||||
Error: errMsg,
|
||||
RequestBody: reqBody,
|
||||
ResponseBody: string(respBody),
|
||||
// Raw wire bodies; empty when the transport did not fill the Capture
|
||||
// (e.g. a provider dialing through a client without the capture hook, or
|
||||
// a call that failed before any HTTP request went out).
|
||||
RawRequest: capt.RawRequest(),
|
||||
RawResponse: capt.RawResponse(),
|
||||
}
|
||||
|
||||
if err := r.pg.InsertLLMRecord(rec); err != nil {
|
||||
|
||||
Reference in new issue
Block a user