From f3058a64c38e9ff6100361ab3d5a901872f208e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=AD=86?= <2685475410@qq.com> Date: Fri, 21 Aug 2026 22:01:41 +0800 Subject: [PATCH] =?UTF-8?q?M0=EF=BC=9A=E7=A5=9E=E8=AF=86=20Go=20=E7=BD=91?= =?UTF-8?q?=E5=85=B3=E9=AA=A8=E6=9E=B6=EF=BC=88IR=20=E5=85=A5=E7=AB=99/SSE?= =?UTF-8?q?=20=E5=87=BA=E7=AB=99/stub=20=E9=80=82=E9=85=8D=E5=99=A8/?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E4=BB=A4=E7=89=8C/=E5=81=A5=E5=BA=B7?= =?UTF-8?q?=E6=A3=80=E6=9F=A5=EF=BC=89=E5=8F=8A=E6=B5=8B=E8=AF=95=E4=B8=8E?= =?UTF-8?q?=E9=95=9C=E5=83=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docker/shenshi/Dockerfile | 15 ++++ shenshi/README.md | 37 +++++++++ shenshi/go.mod | 3 + shenshi/internal/ir/types.go | 66 +++++++++++++++ shenshi/internal/relay/sse.go | 46 +++++++++++ shenshi/internal/route/registry.go | 46 +++++++++++ shenshi/internal/server/server.go | 110 +++++++++++++++++++++++++ shenshi/internal/server/server_test.go | 110 +++++++++++++++++++++++++ shenshi/internal/transform/stub.go | 49 +++++++++++ shenshi/main.go | 32 +++++++ 10 files changed, 514 insertions(+) create mode 100644 docker/shenshi/Dockerfile create mode 100644 shenshi/README.md create mode 100644 shenshi/go.mod create mode 100644 shenshi/internal/ir/types.go create mode 100644 shenshi/internal/relay/sse.go create mode 100644 shenshi/internal/route/registry.go create mode 100644 shenshi/internal/server/server.go create mode 100644 shenshi/internal/server/server_test.go create mode 100644 shenshi/internal/transform/stub.go create mode 100644 shenshi/main.go diff --git a/docker/shenshi/Dockerfile b/docker/shenshi/Dockerfile new file mode 100644 index 0000000..237342b --- /dev/null +++ b/docker/shenshi/Dockerfile @@ -0,0 +1,15 @@ +# 神识（Go 模型网关）构建与运行 +FROM golang:1.25-alpine AS builder +WORKDIR /src +COPY go.mod ./ +RUN go mod download || true +COPY . . +RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/shenshi . + +FROM alpine:3.20 +RUN adduser -D -H -u 10001 shenshi +USER shenshi +COPY --from=builder /out/shenshi /usr/local/bin/shenshi +ENV SHENSHI_PORT=8080 +EXPOSE 8080 +ENTRYPOINT ["shenshi"] diff --git a/shenshi/README.md b/shenshi/README.md new file mode 100644 index 0000000..03f6f77 --- /dev/null +++ b/shenshi/README.md @@ -0,0 +1,37 @@ +# 神识 · 魂镜模型网关（hunjing-shenshi / Shenshi） + +> 文化命名「神识」（契约词典 §2.1）：统一模型能力——Provider/Endpoint 接入、能力槽位、协议转换、路由、流式、BYOK、配额与成本。神识不拥有人物状态。 + +## 当前状态（rewrite/1.0 M0 骨架） + +- 数据面 `POST /v1/chat/completions`：接受灵台内部 IR（`internal/ir`），经路由选适配器，以 SSE 流出标准化增量（`data: ` + `data: [DONE]`）； +- 当前唯一适配器为 `stub` 回声桩（`internal/transform/stub.go`），供端到端骨架验证；云 Provider（openai_compatible/anthropic）与本地元炁端点按 [重写执行计划](../docs/operations/rewrite-plan.md) M4 落地； +- 健康面 `GET /healthz` `GET /readyz`； +- 服务令牌：`SHENSHI_SERVICE_TOKEN` 配置时校验 `Authorization: Bearer`（开发档可空）。 + +## 目录结构 + +```text +shenshi/ + main.go # 入口：配置与路由装配 + internal/ir/types.go # IR 请求/增量（与灵台 app/shenshi_client/types.py 对齐） + internal/transform/ # 出站适配器（Adaptor 接口 + stub） + internal/route/registry.go # 槽位 → 端点候选注册表 + internal/relay/sse.go # SSE 帧写出（X-Accel-Buffering: no） + internal/server/ # HTTP 服务与测试 +``` + +## 本地运行 + +```bash +SHENSHI_PORT=8080 go run . +# 灵台侧配置（backend .env）：SHENSHI_BASE_URL=http://127.0.0.1:8080 +``` + +## 测试 + +```bash +go vet ./... && go test ./... +``` + +设计依据：[神识模型网关](../docs/architecture/shenshi-gateway.md)；错误码段 HJ-2xxx 见[契约词典](../docs/contracts/contract-glossary.md) §15。 diff --git a/shenshi/go.mod b/shenshi/go.mod new file mode 100644 index 0000000..45273d4 --- /dev/null +++ b/shenshi/go.mod @@ -0,0 +1,3 @@ +module hunjing/shenshi + +go 1.25 diff --git a/shenshi/internal/ir/types.go b/shenshi/internal/ir/types.go new file mode 100644 index 0000000..d5a4593 --- /dev/null +++ b/shenshi/internal/ir/types.go @@ -0,0 +1,66 @@ +// Package ir 定义灵台 → 神识的内部中间格式（IR）。 +// 字段与灵台 app/shenshi_client/types.py 逐一对齐，是跨服务契约的 Go 侧表达； +// 契约取值以 docs/contracts/contract-glossary.md 为单一事实源。 +package ir + +import "fmt" + +// Message IR 消息。Tool 消息须携带 ToolCallID，与工具调用配对不拆散。 +type Message struct { + Role string `json:"role"` + Content string `json:"content"` + ToolCallID string `json:"tool_call_id,omitempty"` + Name string `json:"name,omitempty"` +} + +// ChatRequest 版本化模型请求（数据面唯一入站形态）。 +type ChatRequest struct { + IRVersion int `json:"ir_version"` + Slot string `json:"slot"` + Model string `json:"model,omitempty"` + Messages []Message `json:"messages"` + RunID string `json:"run_id"` + TraceID string `json:"trace_id,omitempty"` + CharacterID string `json:"character_id,omitempty"` + Temperature *float64 `json:"temperature,omitempty"` + MaxTokens *int `json:"max_tokens,omitempty"` + Stream bool `json:"stream"` +} + +// Delta 神识标准化流增量（与 Provider 方言无关）。 +type Delta struct { + Delta string `json:"delta,omitempty"` + FinishReason string `json:"finish_reason,omitempty"` + UsagePromptTokens int `json:"usage_prompt_tokens,omitempty"` + UsageCompletionTokens int `json:"usage_completion_tokens,omitempty"` +} + +// SupportedIRVersion 当前支持的 IR 版本；不匹配时返回错误（HJ-2002）。 +const SupportedIRVersion = 1 + +// Validate 校验请求形态：版本、必填字段与流式开关。 +func (r *ChatRequest) Validate() error { + if r.IRVersion != SupportedIRVersion { + return fmt.Errorf("ir_version %d 不受支持（当前 %d）", r.IRVersion, SupportedIRVersion) + } + if r.RunID == "" { + return fmt.Errorf("run_id 不能为空") + } + if len(r.Messages) == 0 { + return fmt.Errorf("messages 不能为空") + } + if !r.Stream { + return fmt.Errorf("数据面当前只接受流式请求（stream=true）") + } + return nil +} + +// LastUserMessage 返回最后一条用户消息内容（stub 适配器回声源）。 +func (r *ChatRequest) LastUserMessage() string { + for i := len(r.Messages) - 1; i >= 0; i-- { + if r.Messages[i].Role == "user" { + return r.Messages[i].Content + } + } + return "" +} diff --git a/shenshi/internal/relay/sse.go b/shenshi/internal/relay/sse.go new file mode 100644 index 0000000..b09390f --- /dev/null +++ b/shenshi/internal/relay/sse.go @@ -0,0 +1,46 @@ +// Package relay 负责流式转发：把适配器增量以 SSE 帧写出。 +// 转发红线（shenshi-gateway.md §5）：X-Accel-Buffering 防代理缓冲；帧即写即刷。 +package relay + +import ( + "encoding/json" + "fmt" + "net/http" + + "hunjing/shenshi/internal/ir" +) + +// SSEHeaders 设置流式响应头（在写第一帧前调用）。 +func SSEHeaders(w http.ResponseWriter) { + w.Header().Set("Content-Type", "text/event-stream; charset=utf-8") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("X-Accel-Buffering", "no") +} + +// WriteDelta 写一帧增量事件（data: ）。 +func WriteDelta(w http.ResponseWriter, d ir.Delta) error { + body, err := json.Marshal(d) + if err != nil { + return fmt.Errorf("序列化增量失败: %w", err) + } + _, err = fmt.Fprintf(w, "data: %s\n\n", body) + if err != nil { + return err + } + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + return nil +} + +// WriteDone 写终止帧（data: [DONE]）。 +func WriteDone(w http.ResponseWriter) error { + _, err := fmt.Fprint(w, "data: [DONE]\n\n") + if err != nil { + return err + } + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + return nil +} diff --git a/shenshi/internal/route/registry.go b/shenshi/internal/route/registry.go new file mode 100644 index 0000000..959581d --- /dev/null +++ b/shenshi/internal/route/registry.go @@ -0,0 +1,46 @@ +// Package route 维护逻辑模型/能力槽位到候选端点的路由注册表。 +// M0 为单一 stub 端点的静态表；候选迭代、健康冷却与故障切换按 +// docs/architecture/shenshi-gateway.md §4 于 M4 落地。 +package route + +import ( + "fmt" + + "hunjing/shenshi/internal/transform" +) + +// Endpoint 一个可路由的出站端点：类型 + 适配器实例。 +type Endpoint struct { + Type string + Adaptor transform.Adaptor + Priority int +} + +// Registry 槽位 → 候选端点列表（按 Priority 升序尝试）。 +type Registry struct { + bySlot map[string][]Endpoint +} + +// New 构建注册表并登记默认端点。 +func New() *Registry { + r := &Registry{bySlot: make(map[string][]Endpoint)} + r.Register("chat", Endpoint{Type: "stub", Adaptor: transform.StubAdaptor{}, Priority: 100}) + return r +} + +// Register 登记端点到指定能力槽位（契约词典 §14.1 model_slot 取值）。 +func (r *Registry) Register(slot string, ep Endpoint) { + r.bySlot[slot] = append(r.bySlot[slot], ep) +} + +// Resolve 返回槽位的候选端点；无候选即错误（HJ-2001 语义）。 +func (r *Registry) Resolve(slot string) ([]Endpoint, error) { + if slot == "" { + slot = "chat" + } + eps := r.bySlot[slot] + if len(eps) == 0 { + return nil, fmt.Errorf("槽位 %s 无可用端点", slot) + } + return eps, nil +} diff --git a/shenshi/internal/server/server.go b/shenshi/internal/server/server.go new file mode 100644 index 0000000..c21f3c3 --- /dev/null +++ b/shenshi/internal/server/server.go @@ -0,0 +1,110 @@ +// Package server 神识 HTTP 服务：数据面（IR 入站 → SSE 出站）与健康面。 +package server + +import ( + "encoding/json" + "net/http" + + "hunjing/shenshi/internal/ir" + "hunjing/shenshi/internal/relay" + "hunjing/shenshi/internal/route" +) + +// Config 服务配置。 +type Config struct { + ServiceToken string // 灵台→神识服务令牌；为空时开发档不校验 +} + +// Server 神识 HTTP 服务。 +type Server struct { + cfg Config + routes *route.Registry + mux *http.ServeMux +} + +// New 构建服务并注册路由。 +func New(cfg Config, routes *route.Registry) *Server { + s := &Server{cfg: cfg, routes: routes, mux: http.NewServeMux()} + s.mux.HandleFunc("GET /healthz", s.handleHealth) + s.mux.HandleFunc("GET /readyz", s.handleHealth) + s.mux.Handle("POST /v1/chat/completions", s.auth(http.HandlerFunc(s.handleChat))) + return s +} + +// ServeHTTP 实现 http.Handler。 +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.mux.ServeHTTP(w, r) } + +// auth 服务令牌校验中间件；未配置令牌时放行（本地开发档）。 +func (s *Server) auth(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if s.cfg.ServiceToken == "" { + next.ServeHTTP(w, r) + return + } + credential := r.Header.Get("Authorization") + if credential != "Bearer "+s.cfg.ServiceToken { + writeError(w, http.StatusUnauthorized, "HJ-2003", "服务令牌无效") + return + } + next.ServeHTTP(w, r) + }) +} + +// handleChat 数据面：解析 IR → 路由 → 适配器流出 SSE。 +func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { + var req ir.ChatRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "HJ-2002", "IR 请求解析失败: "+err.Error()) + return + } + if err := req.Validate(); err != nil { + writeError(w, http.StatusBadRequest, "HJ-2002", err.Error()) + return + } + endpoints, err := s.routes.Resolve(req.Slot) + if err != nil { + writeError(w, http.StatusServiceUnavailable, "HJ-2001", err.Error()) + return + } + + relay.SSEHeaders(w) + w.WriteHeader(http.StatusOK) + + deltas := make(chan ir.Delta, 16) + streamErr := make(chan error, 1) + go func() { + defer close(deltas) + streamErr <- endpoints[0].Adaptor.Stream(r.Context(), &req, deltas) + }() + for d := range deltas { + if err := relay.WriteDelta(w, d); err != nil { + return // 客户端断开：取消上游由 r.Context() 传播 + } + } + if err := <-streamErr; err != nil { + // 流已开始，只能如实中断：写一条错误帧后终止（不伪装 [DONE]） + _ = relay.WriteDelta(w, ir.Delta{FinishReason: "error", Delta: ""}) + _ = relay.WriteDone(w) + return + } + if err := relay.WriteDone(w); err != nil { + return + } +} + +// handleHealth 健康面。 +func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ok"}`)) +} + +// writeError 统一错误结构（API 契约 §5）。 +func writeError(w http.ResponseWriter, status int, code, message string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + body := map[string]any{ + "error": map[string]string{"code": code, "message": message}, + } + _ = json.NewEncoder(w).Encode(body) +} diff --git a/shenshi/internal/server/server_test.go b/shenshi/internal/server/server_test.go new file mode 100644 index 0000000..938f1eb --- /dev/null +++ b/shenshi/internal/server/server_test.go @@ -0,0 +1,110 @@ +package server + +import ( + "bufio" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "hunjing/shenshi/internal/route" +) + +func newTestServer(token string) *httptest.Server { + srv := New(Config{ServiceToken: token}, route.New()) + return httptest.NewServer(srv) +} + +func chatBody() string { + return `{"ir_version":1,"slot":"chat","messages":[` + + `{"role":"system","content":"契约占位"},{"role":"user","content":"你好"}],` + + `"run_id":"run_test","stream":true}` +} + +func post(t *testing.T, ts *httptest.Server, token, body string) (*http.Response, string) { + t.Helper() + req, err := http.NewRequest(http.MethodPost, ts.URL+"/v1/chat/completions", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + resp, err := ts.Client().Do(req) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = resp.Body.Close() }) + var sb strings.Builder + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + sb.WriteString(scanner.Text()) + sb.WriteString("\n") + } + return resp, sb.String() +} + +func TestChatCompletionsStreamsEchoAndDone(t *testing.T) { + ts := newTestServer("") + resp, body := post(t, ts, "", chatBody()) + if resp.StatusCode != http.StatusOK { + t.Fatalf("状态码 = %d", resp.StatusCode) + } + if got := resp.Header.Get("X-Accel-Buffering"); got != "no" { + t.Fatalf("X-Accel-Buffering = %q", got) + } + if !strings.Contains(body, `"delta":"【神识回声桩】"`) && !strings.Contains(body, "神识回声桩") { + t.Fatalf("缺少回声增量: %s", body) + } + if !strings.Contains(body, `"finish_reason":"stop"`) { + t.Fatalf("缺少终帧用量: %s", body) + } + if !strings.Contains(body, "data: [DONE]") { + t.Fatalf("缺少 [DONE]: %s", body) + } +} + +func TestServiceTokenRejectsInvalidCredential(t *testing.T) { + ts := newTestServer("secret-token") + resp, body := post(t, ts, "wrong", chatBody()) + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("状态码 = %d", resp.StatusCode) + } + if !strings.Contains(body, "HJ-2003") { + t.Fatalf("缺少错误码: %s", body) + } +} + +func TestInvalidIRRejected(t *testing.T) { + ts := newTestServer("") + resp, body := post(t, ts, "", `{"ir_version":99,"messages":[],"run_id":"","stream":true}`) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("状态码 = %d", resp.StatusCode) + } + if !strings.Contains(body, "HJ-2002") { + t.Fatalf("缺少错误码: %s", body) + } +} + +func TestUnknownSlotNoEndpoint(t *testing.T) { + ts := newTestServer("") + resp, body := post(t, ts, "", `{"ir_version":1,"slot":"vision","messages":[{"role":"user","content":"x"}],"run_id":"run_t","stream":true}`) + if resp.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("状态码 = %d", resp.StatusCode) + } + if !strings.Contains(body, "HJ-2001") { + t.Fatalf("缺少错误码: %s", body) + } +} + +func TestHealthz(t *testing.T) { + ts := newTestServer("") + resp, err := ts.Client().Get(ts.URL + "/healthz") + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Fatalf("状态码 = %d", resp.StatusCode) + } +} diff --git a/shenshi/internal/transform/stub.go b/shenshi/internal/transform/stub.go new file mode 100644 index 0000000..2a0a3be --- /dev/null +++ b/shenshi/internal/transform/stub.go @@ -0,0 +1,49 @@ +// Package transform 定义出站适配器：IR → Provider 方言的转换层。 +// M0 只含回声 stub 适配器；云 Provider adaptor（openai_compatible/anthropic）与 +// 本地元炁端点按 rewrite-plan M4 落地（docs/architecture/shenshi-gateway.md §5）。 +package transform + +import ( + "context" + + "hunjing/shenshi/internal/ir" +) + +// Adaptor 出站适配器契约：消费 IR 请求，产出标准化增量流。 +type Adaptor interface { + // Stream 产出增量；context 取消时须及时返回。 + Stream(ctx context.Context, req *ir.ChatRequest, out chan<- ir.Delta) error + // Name 适配器稳定标识（endpoint.type 取值，契约词典 §14.2）。 + Name() string +} + +// chunkSize stub 回声切分粒度（与灵台 EchoShenshi 对齐，便于联调对比）。 +const chunkSize = 8 + +// StubAdaptor 回声适配器：把最后一条用户消息加前缀后按固定粒度回流。 +// 仅用于 M0 骨架端到端验证，不模拟任何 Provider 行为。 +type StubAdaptor struct{} + +// Name 实现 Adaptor。 +func (StubAdaptor) Name() string { return "stub" } + +// Stream 实现 Adaptor：逐块产出增量并在末帧携带用量。 +func (StubAdaptor) Stream(ctx context.Context, req *ir.ChatRequest, out chan<- ir.Delta) error { + prompt := req.LastUserMessage() + runes := []rune("【神识回声桩】" + prompt) // 按 rune 切分，与灵台回声桩的码点语义对齐 + for i := 0; i < len(runes); i += chunkSize { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + end := min(i+chunkSize, len(runes)) + out <- ir.Delta{Delta: string(runes[i:end])} + } + out <- ir.Delta{ + FinishReason: "stop", + UsagePromptTokens: max(len([]rune(prompt))/4, 1), + UsageCompletionTokens: max(len(runes)/4, 1), + } + return nil +} diff --git a/shenshi/main.go b/shenshi/main.go new file mode 100644 index 0000000..7f91e94 --- /dev/null +++ b/shenshi/main.go @@ -0,0 +1,32 @@ +// 神识（Shenshi）入口：魂镜模型网关，rewrite/1.0 M0 骨架。 +// 数据面 POST /v1/chat/completions（IR 入站，SSE 出站）；健康面 /healthz /readyz。 +package main + +import ( + "log" + "net/http" + "os" + + "hunjing/shenshi/internal/route" + "hunjing/shenshi/internal/server" +) + +func envOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +func main() { + port := envOr("SHENSHI_PORT", "8080") + srv := server.New( + server.Config{ServiceToken: os.Getenv("SHENSHI_SERVICE_TOKEN")}, + route.New(), + ) + addr := ":" + port + log.Printf("神识就绪：http://0.0.0.0%s（stub 端点，等待 M4 接入真实 Provider）", addr) + if err := http.ListenAndServe(addr, srv); err != nil { + log.Fatal(err) + } +}