Provider abstraction và adapter Anthropic

Tổng quan

Định nghĩa mô hình LLM chuẩn hóa của hango — Message, Block, Event, StopReason, Usage — và interface Provider streaming một method, rồi cài nó trên anthropic-sdk-go v1.74. Mô hình chuẩn hóa được thiết kế ngay bây giờ để adapter OpenAI ở Phase 12 chỉ cần tổng hợp phần khung block; làm sai đường nối này là sai lầm đắt nhất có thể mắc trong roadmap. Không có công việc về database: giai đoạn này hoàn toàn là code thư viện cộng một lệnh CLI smoke.

Yêu cầu

Chức năng

• provider.Provider có một method streaming duy nhất trả về iter.Seq[Event]. Bên gọi không bao giờ duyệt lại một sequence để “gom lại sau”: họ đẩy từng event vào một Accumulator có trạng thái trong khi chuyển tiếp nó tới sink. Accumulator có Push(Event) error, Message() provider.Message, Partial() provider.Message, Usage() Usage, Stop() StopReason. Khi có lỗi hoặc context bị hủy, bên gọi giữ message dở dang từ Partial() cùng với lỗi — đường hủy ở Phase 6 và đường compaction ở Phase 10 đều phụ thuộc việc phần dở dang đó có sẵn.
• Các block gồm text, thinking (kèm signature), tool_use, tool_result; []Block marshal thành JSON ổn định để lưu jsonb ở Phase 4.
• Các thinking block đi vòng nguyên vẹn (text + signature) về lại provider ở lượt kế tiếp.
• Mọi stop reason của Anthropic đều ánh xạ sang các hằng số của hango, gồm cả model_context_window_exceeded. pause_turn ánh xạ sang StopReason PauseTurn, mà adapter lập tức đưa ra thành ErrUnsupportedStop: hango không đăng ký server tool nào của Anthropic, nên một pause_turn chỉ có thể có nghĩa là provider đã làm điều gì đó mà hango không biểu diễn được trong mô hình block chuẩn hóa. Không có logic resume nào ở bất kỳ đâu trong roadmap. Thêm server tool (và việc chuyển tiếp raw block mờ mà chúng đòi hỏi) là một việc theo dõi trong plan.md.
• Các lỗi có kiểu: ErrRateLimited (mang RetryAfter), ErrOverloaded, ErrContextOverflow, ErrAuth, ErrInvalidRequest, ErrProvider, ErrUnsupportedStop (mang chuỗi stop reason thô).
• Retry với exponential backoff và jitter khi gặp lỗi 429/5xx/mạng, tôn trọng Retry-After; số lần retry có giới hạn và có thể hủy theo context.
• Usage mang các token input, output, reasoning, cache-read và cache-write, với cache-write được tách thành CacheWrite5m và CacheWrite1h. Anthropic tính giá hai TTL khác nhau; một trường cache-write duy nhất khiến việc tính giá đúng ở Phase 13 bất khả, và việc tách này phải có trong mô hình chuẩn hóa ngay từ đầu.
• hango probe --provider anthropic --prompt "..." stream một completion ra stdout để kiểm tra thủ công.
• Langfuse chỉ nhận các span LLM và agent. Một processor filter/langfuse_genai_only trên pipeline traces/langfuse bỏ mọi span không có attribute gen_ai.operation.name, nên các span HTTP, health-check và database chỉ nằm ở Tempo (quyết định ngày 2026-10-14). Phase 1 gửi mọi span tới Langfuse thuần túy để chứng minh đường xuất dữ liệu. Các span invoke_agent và execute_tool của Phase 5 cũng mang gen_ai.operation.name, nên chúng qua cùng bộ lọc. Langfuse hiển thị các span agent không có span HTTP cha của chúng, đó là góc nhìn chủ ý.

Phi chức năng

• internal/provider không import package hango nào ngoài stdlib; nó không được import store (registry nhận biết tenant ở Phase 12 tiêm credential vào, chứ không đảo chiều dependency).
• Test mang tính xác định: một provider fake trong package cùng các test golden-event trên các chuỗi event SDK đã ghi lại. Không có mạng trong go test.
• testing/synctest cho test thời gian retry/backoff.

Kiến trúc

// internal/provider/provider.go
type Provider interface {
	Name() string
	DefaultModel() string
	Stream(ctx context.Context, req Request) (iter.Seq[Event], error)
}

type Request struct {
	Model       string
	System      []Block          // text blocks; cache breakpoints set via Block.CacheControl
	Messages    []Message
	Tools       []ToolDef
	MaxTokens   int
	Temperature *float64
	Thinking    *ThinkingConfig  // nil = disabled
	Metadata    map[string]string
}

// internal/provider/message.go
type Role string // "user" | "assistant"

type Message struct {
	Role   Role    `json:"role"`
	Blocks []Block `json:"blocks"`
}

type Block struct { // exactly one non-nil
	Text         *TextBlock       `json:"text,omitempty"`
	Thinking     *ThinkingBlock   `json:"thinking,omitempty"`
	ToolUse      *ToolUseBlock    `json:"tool_use,omitempty"`
	ToolResult   *ToolResultBlock `json:"tool_result,omitempty"`
	CacheControl *CacheControl    `json:"cache_control,omitempty"`
}

type ThinkingBlock struct {
	Text      string `json:"text"`
	Signature string `json:"signature"` // echoed back verbatim or tool use breaks
	Redacted  []byte `json:"redacted,omitempty"`
}

type ToolUseBlock struct {
	ID    string          `json:"id"`
	Name  string          `json:"name"`
	Input json.RawMessage `json:"input"`
}

type ToolResultBlock struct {
	ToolUseID string  `json:"tool_use_id"`
	Content   []Block `json:"content"`
	IsError   bool    `json:"is_error,omitempty"`
}

type StopReason uint8 // Unknown, EndTurn, MaxTokens, StopSequence, ToolUse, Refusal, PauseTurn, ContextExceeded
// PauseTurn is terminal for hango: it is surfaced as ErrUnsupportedStop, never resumed. // Red team: D4

// internal/provider/event.go
type EventKind uint8 // MessageStart, BlockStart, TextDelta, ThinkingDelta, SignatureDelta,
                     // ToolInputDelta, BlockStop, MessageStop, Error

type Event struct {
	Kind  EventKind  `json:"kind"`
	Index int        `json:"index,omitempty"` // block index; OpenAI maps tool_call delta Index here
	Delta string     `json:"delta,omitempty"`
	Block *Block     `json:"block,omitempty"` // BlockStart only; ToolUse carries ID+Name, empty Input
	Stop  StopReason `json:"stop,omitempty"`
	Usage *Usage     `json:"usage,omitempty"`
	Err   error      `json:"-"`
}

// Red team: F7 — stateful accumulator, not a terminal fold over the sequence, so a caller can
// stream each event onward and still hold a consistent partial message at any point.
type Accumulator struct{ /* ordered in-progress blocks by index, usage, stop, err */ }

func NewAccumulator() *Accumulator
func (a *Accumulator) Push(ev Event) error   // returns the terminal error carried by an Error event
func (a *Accumulator) Message() Message      // finalized; valid after MessageStop
func (a *Accumulator) Partial() Message      // best-effort snapshot at any point, incl. after an error
func (a *Accumulator) Usage() Usage
func (a *Accumulator) Stop() StopReason

Giai đoạn này không có bảng nào.

Instrumentation

• Một span cho mỗi lời gọi provider: tên chat {model}, kind CLIENT, được tạo bên trong adapter để mọi cài đặt provider đều có nó miễn phí thông qua một helper dùng chung trong internal/provider/trace.go.
• Attribute: gen_ai.operation.name=chat, gen_ai.provider.name=anthropic và gen_ai.system=anthropic (Langfuse vẫn ghi tài liệu tên cũ, nên phát cả hai), gen_ai.request.model, gen_ai.response.model, gen_ai.request.max_tokens, gen_ai.request.temperature, gen_ai.request.stream=true, gen_ai.response.id, gen_ai.response.finish_reasons, gen_ai.response.time_to_first_chunk, gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, gen_ai.usage.cache_read.input_tokens, gen_ai.usage.cache_write.input_tokens, gen_ai.usage.reasoning.output_tokens.
• Attribute của Langfuse: langfuse.observation.type=generation, langfuse.observation.model.name, langfuse.observation.completion_start_time, langfuse.observation.usage_details (JSON tường minh để việc tính giá cache token không phụ thuộc vào việc Langfuse suy luận tên key semconv, và mang cả phần tách cache-write 5m/1h), và langfuse.observation.input/langfuse.observation.output chỉ được ghi khi HANGO_TRACE_LLM_CONTENT=true và trace_llm_content của tenant là true. Cả hai cổng được kiểm tra tại đúng một nơi đặt content attribute; mọi key được ghi ở đó đều có trong internal/telemetry/content_attrs.go, và collector loại bỏ danh sách đó trên cả nhánh Langfuse cho các tenant đã opt-out.
• Retry: mỗi lần thử là một span event provider.retry với attempt, delay_ms, error.type, http.response.status_code, retry_after_s.
• Metric: histogram gen_ai.client.token.usage (attribute gen_ai.token.type = input|output|cache_read|cache_write, gen_ai.request.model, gen_ai.provider.name), gen_ai.client.operation.duration, gen_ai.client.operation.time_to_first_chunk, gen_ai.client.operation.time_per_output_chunk, counter hango.provider.errors với error.type.
• Lỗi: span.RecordError cùng span.SetStatus(codes.Error) và langfuse.observation.level=ERROR với langfuse.observation.status_message.

Các file liên quan

Tạo mới

• internal/provider/provider.go — Provider, Request, ToolDef, ThinkingConfig, CacheControl.
• internal/provider/message.go — các kiểu message/block cùng kiểm tra trong MarshalJSON/UnmarshalJSON rằng đúng một trường block được đặt.
• internal/provider/event.go — các kiểu event và Accumulator.
• internal/provider/errors.go — các lỗi có kiểu và IsRetryable(err) bool.
• internal/provider/trace.go — helper span dùng chung cho mọi adapter (bắt đầu, đặt attribute, đo thời gian chunk đầu, ghi usage, trạng thái lỗi) cộng các metric instrument.
• internal/provider/retry.go — Do(ctx, policy, fn) với exponential backoff, jitter và Retry-After.
• internal/provider/fake.go — provider theo kịch bản được dùng bởi test vòng lặp của mọi giai đoạn sau.
• internal/provider/anthropic/provider.go — dựng adapter và Stream.
• internal/provider/anthropic/convert.go — Request của hango sang tham số SDK, block của SDK sang block của hango.
• internal/provider/anthropic/events.go — event union của SDK sang Event của hango.
• internal/provider/anthropic/testdata/*.json — các chuỗi event đã ghi (text, tool_use, thinking, overflow, pause_turn cho khẳng định ErrUnsupportedStop).
• internal/provider/*_test.go, internal/provider/anthropic/*_test.go.
• cmd/hango/probe.go — subcommand probe.
• docs/learning/03-provider-abstraction.md.

Sửa đổi

• internal/config/config.go — HANGO_ANTHROPIC_API_KEY, HANGO_ANTHROPIC_BASE_URL, HANGO_DEFAULT_MODEL, HANGO_PROVIDER_MAX_RETRIES, HANGO_PROVIDER_TIMEOUT.
• internal/config/config.go — thêm HANGO_TRACE_LLM_CONTENT (mặc định false).
• internal/app/app.go — dựng provider mặc định; bỏ qua khi không có key nào được đặt.
• deploy/otelcol/config.yaml và test generator của nó trong internal/telemetry — thêm filter/langfuse_genai_only vào pipeline traces/langfuse.
• docs/system-architecture.md, docs/learning/README.md, docs/observability-cloud-setup.md.

Các bước triển khai

1. Thêm github.com/anthropics/anthropic-sdk-go v1.74.0.
2. Viết message.go. Thêm helper Block.Kind() string và Validate() error bắt buộc đúng một trường được điền. Viết test JSON round-trip cho mọi kiểu block, gồm cả một thinking block có signature.
3. Viết event.go với Accumulator: duy trì một slice có thứ tự các block đang dở dang được đánh index theo Index; BlockStart gieo mầm, các delta nối thêm (các delta input của tool nối vào một buffer được parse ở BlockStop), MessageStop chốt. Push trả về lỗi của một event Error và giữ lại mọi thứ đã tích lũy, để Partial() có ý nghĩa sau một thất bại hoặc một lần hủy. Test: đẩy một chuỗi bị cắt ngắn, khẳng định Partial() chứa các block đã hoàn tất và đoạn text đang dở.
4. Viết errors.go và retry.go. IsRetryable bao gồm 429, 5xx, timeout của net.Error và io.ErrUnexpectedEOF. Backoff cơ sở 500ms, hệ số 2, trần 30s, full jitter, tôn trọng Retry-After khi nó lớn hơn. Test bằng testing/synctest.
5. Viết anthropic/convert.go: ánh xạ Request.System sang SystemBlocks với CacheControl, message sang các MessageParam, định nghĩa tool sang ToolUnionParam. Các thinking block phải được phát lại dưới dạng ThinkingBlockParam với signature gốc; thinking bị che được phát lại dưới dạng RedactedThinkingBlockParam.
6. Viết anthropic/events.go: vòng lặp stream.Next()/Current(), type switch event.AsAny() tạo ra các event của hango. Ánh xạ ContentBlockStartEvent thành BlockStart (tool_use mang ID và Name), TextDelta/ThinkingDelta/SignatureDelta/InputJSONDelta thành các loại delta của chúng, ContentBlockStopEvent thành BlockStop, MessageDeltaEvent thành stop reason + usage, MessageStopEvent thành MessageStop. Đưa stream.Err() ra thành một event Error kết thúc để sequence vẫn chỉ có một kiểu giá trị.
7. Ánh xạ các stop reason: end_turn, max_tokens, stop_sequence, tool_use, refusal, pause_turn, model_context_window_exceeded sang các hằng số của hango; model_context_window_exceeded còn đưa ra ErrContextOverflow để Phase 10 kích hoạt compaction, và pause_turn đưa ra ErrUnsupportedStop cùng lý do thô để vòng lặp làm run thất bại với một thông báo tường minh, grep được thay vì âm thầm cắt cụt một lượt.
8. Viết anthropic/provider.go: dựng SDK client với base URL, API key, timeout; tắt retry ở tầng SDK và dùng retry.Do của hango để chính sách backoff chỉ có một cài đặt. Stream trả về một iter.Seq[Event] mà yield của nó tôn trọng ctx.Done() và luôn đóng SDK stream khi thoát.
9. Viết fake.go: một provider dựng từ một kịch bản [][]Event trả về chuỗi kế tiếp cho mỗi lời gọi, với các hook để tiêm lỗi. Đây là test double mà các giai đoạn vòng lặp sẽ dùng.
10. Viết test golden giải mã testdata/*.json thành các event union của SDK và khẳng định chuỗi event hango được tạo ra cùng Message được tích lũy.
11. Thêm cmd/hango/probe.go và kiểm tra thủ công với API thật (loại khỏi go test).
12. Viết internal/provider/trace.go và nối nó vào Stream của adapter Anthropic: bắt đầu span trước request, ghi time_to_first_chunk ở delta đầu tiên, đặt các attribute usage và finish-reason ở MessageStop, kết thúc span trong một defer. Thêm một test in-memory span-recorder khẳng định đầy đủ tập attribute cho một stream theo kịch bản, và khẳng định các content attribute vắng mặt khi HANGO_TRACE_LLM_CONTENT=false.
13. Chạy một probe Langfuse thực nghiệm 10 phút (`hango probe --trace` với Langfuse Cloud) để chốt ba hành vi mà nghiên cứu đã đánh dấu CHƯA XÁC MINH: (a) `gen_ai.input.messages`/`gen_ai.output.messages` có hiển thị thành input/output của observation hay chỉ `gen_ai.prompt`/`gen_ai.completion`, (b) engine tính giá của Langfuse tính cho cache token của Anthropic theo các key usage nào, (c) `gen_ai.operation.name` một mình có suy ra được kiểu observation không. Ghi kết quả vào learning doc. Phương án dự phòng đã định trước bất kể kết quả: luôn đặt các attribute `langfuse.*` một cách tường minh.
14. Thêm filter/langfuse_genai_only vào pipeline traces/langfuse, sau filter/langfuse_disabled, với điều kiện span attributes["gen_ai.operation.name"] == nil. Mở rộng test config của collector để khẳng định processor có mặt trên nhánh Langfuse và vắng mặt khỏi các nhánh lgtm và Grafana Cloud. Xác minh thủ công: sau curl /readyz và một lần hango probe --trace, Langfuse chỉ hiển thị generation chat {model}, trong khi Tempo hiển thị cả hai trace.
15. Viết learning doc, gồm một mô tả vẽ tay của giao thức SSE thô của Messages API.

Dàn ý tài liệu học

docs/learning/03-provider-abstraction.md phải có:

• Khái niệm: vì sao một harness cần mô hình message riêng thay vì truyền các kiểu của SDK đi khắp nơi.
• Giao thức SSE thô của Anthropic Messages API, được giải thích dù SDK đã che đi: khung text/event-stream, các dòng event:/data:, chuỗi message_start → content_block_start → content_block_delta* → content_block_stop → message_delta → message_stop, các event ping, và việc tích lũy partial_json có nghĩa là gì. Kèm một bản ghi event nguyên văn có chú thích.
• Các lựa chọn thiết kế: một method streaming với iter.Seq[Event], lỗi được mang như một event kết thúc, khung block là dạng chuẩn hóa, một Accumulator có trạng thái (không phải một phép fold kết thúc) để streaming và tích lũy chia sẻ một lượt duyệt và một message dở dang sống sót qua thất bại, retry do hango sở hữu thay vì SDK.
• Vì sao `pause_turn` không thể xảy ra với hango và do đó là một lỗi chứ không phải một vòng lặp resume: nó tồn tại cho server tool của Anthropic, mà hango không đăng ký cái nào.
• Đánh đổi: SDK so với wire client viết tay (chi phí giải mã union, tích lũy, signature của thinking); interface khả năng tùy chọn so với các trường khả năng.
• Thinking block: vì sao signature phải được trả lại nguyên vẹn và điều gì hỏng nếu không.
• So sánh với goclaw: Provider bốn method (Chat/ChatStream/DefaultModel/Name), Options map[string]any, RawAssistantContent json.RawMessage cho việc trả lại thinking, SSEScanner và RetryDo dùng chung, các interface ThinkingCapable tùy chọn.
• Cách debug tầng này: trong Langfuse, mở observation generation để xem model, tham số, finish reason, token usage và (khi bật content tracing) chính xác prompt và completion; trong Grafana, dùng các histogram gen_ai.client.operation.duration và token-usage theo từng model, và đọc các span event provider.retry trong Tempo để biết một lời gọi chậm là do backoff rate-limit hay do model thật sự chậm. Kết quả của probe ánh xạ Langfuse, với tên attribute chính xác đã hoạt động.
• So sánh với deepagents: nó ủy thác hoàn toàn cho lớp trừu tượng model của LangChain và chỉ thấy các message đã chuẩn hóa — hango phải tự xây thứ mà LangChain cung cấp.
• Vì sao hango đi khác: một method streaming (non-streaming là một phép fold), block có kiểu thay vì trả lại JSON thô, stop reason là một enum tường minh gồm cả trigger cho compaction.
• Bài tập: thêm một kiểu block mới; làm cho khoảng trống khung của OpenAI trở nên cụ thể bằng cách phác thảo cách BlockStart sẽ được tổng hợp. Tùy chọn: viết một vòng lặp think-act-observe 30 dòng dùng một lần trong một file _test.go trên provider.Fake và một tool, để thấy hình dạng của vòng lặp trước khi Phase 5 xây phiên bản production. Không có code sản phẩm.
• Ngân sách thời gian: 12h triển khai + 3h viết learning doc (bản ghi giao thức wire là phần tốn kém nhất).

Tiêu chí thành công

• go build ./... && go vet ./... && go test ./... pass khi không có truy cập mạng.
• Test golden bao phủ các chuỗi text-only, tool_use, thinking-with-signature và model_context_window_exceeded; fixture pause_turn khẳng định ErrUnsupportedStop được trả về và không có nỗ lực resume nào.
• Accumulator trên một stream tool_use đã ghi cho ra một Message có ToolUseBlock.Input là JSON hợp lệ khớp fixture, và Partial() trên một stream bị cắt trả về các block đã hoàn tất cho tới lúc đó.
• Usage mang CacheWrite5m và CacheWrite1h tách biệt từ một fixture dùng cả hai TTL.
• Một thinking block được giải mã và mã hóa lại qua convert.go giữ nguyên signature từng byte (test khẳng định).
• Test retry dưới testing/synctest cho thấy số lần thử có giới hạn, Retry-After được tôn trọng, và trả về ngay khi context bị hủy.
• hango probe --prompt "say hi" stream token với API thật (kiểm tra thủ công, ghi trong learning doc).
• Một span CLIENT chat {model} với đầy đủ tập attribute hiển thị trong Tempo, và cùng lời gọi đó hiển thị thành một generation trong Langfuse với model, usage và finish reason được điền.
• Langfuse chỉ nhận các span mang gen_ai.operation.name. Một request /readyz hiện trong Tempo nhưng không hiện trong Langfuse, và test config của collector khẳng định filter/langfuse_genai_only chỉ nằm trên nhánh Langfuse.
• Với HANGO_TRACE_LLM_CONTENT=false, không có văn bản prompt hay completion nào xuất hiện trên bất kỳ span nào trong Tempo hay Langfuse (kiểm tra thủ công cộng test span-recorder).
• Với HANGO_TRACE_LLM_CONTENT=true nhưng trace_llm_content=false của tenant, span không mang content attribute nào cả (test span-recorder), và mọi content key mà adapter có thể ghi đều có mặt trong content_attrs.go (test liệt kê chúng).

Đánh giá rủi ro

Rollback

Một package thuần cộng thêm cùng một khối cấu hình. Hoàn tác nghĩa là xóa internal/provider và subcommand probe; các Phase 1 và 2 không bị đụng tới vì internal/app chỉ dựng provider khi HANGO_ANTHROPIC_API_KEY được đặt.