From 4b731b5ac0cba47fde794727713f493a3e757a33 Mon Sep 17 00:00:00 2001 From: cfy666 <3087823110@qq.com> Date: Fri, 19 Jun 2026 21:58:17 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20Eino=20Graph=20?= =?UTF-8?q?=E6=9E=84=E5=BB=BA=E4=B8=8E=20Orchestrator=20=E9=80=82=E9=85=8D?= =?UTF-8?q?=E5=99=A8=EF=BC=8C=E5=88=87=E6=8D=A2=20main.go?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - graph.go: 构建 Graph 拓扑 START→STT→History→ChatModel→Splitter→TTS→Done→END - 创建 eino-ext ChatModel 对接 DashScope OpenAI 兼容接口 - 统一使用值类型（PipelineInput/PipelineOutput） - Callback 在运行时通过 Stream option 传入 - adapter.go: EinoOrchestrator 实现 orchestrator.Orchestrator 接口 - 解码 base64 音频/图片，注入 context 值 - 调用 Graph.Stream() 触发惰性执行并消费输出 - 追加用户/助手消息到历史 - main.go: 移除旧 llmService + orchestrator.New() 替换为 eino.NewPipelineGraph() + eino.NewEinoOrchestrator() - 各节点统一使用值类型，State 传递请求元数据 Co-Authored-By: Claude --- backend/cmd/server/main.go | 14 +-- backend/go.mod | 4 +- backend/go.sum | 22 ++++ backend/internal/eino/adapter.go | 165 +++++++++++++++++++++++++ backend/internal/eino/graph.go | 113 +++++++++++++++++ backend/internal/eino/nodes_done.go | 80 ++++-------- backend/internal/eino/nodes_history.go | 61 ++++----- backend/internal/eino/nodes_stt.go | 23 +++- backend/internal/eino/state.go | 11 +- 9 files changed, 395 insertions(+), 98 deletions(-) create mode 100644 backend/internal/eino/adapter.go create mode 100644 backend/internal/eino/graph.go diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index c266172..95fd1e4 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -13,12 +13,11 @@ import ( "github.com/hhs/camtalk/internal/api" "github.com/hhs/camtalk/internal/auth" - "github.com/hhs/camtalk/internal/ai/llm" "github.com/hhs/camtalk/internal/ai/stt" "github.com/hhs/camtalk/internal/ai/tts" "github.com/hhs/camtalk/internal/config" + eino "github.com/hhs/camtalk/internal/eino" "github.com/hhs/camtalk/internal/logger" - "github.com/hhs/camtalk/internal/orchestrator" "github.com/hhs/camtalk/internal/session" "github.com/hhs/camtalk/internal/store" "github.com/hhs/camtalk/internal/ws" @@ -116,9 +115,6 @@ func main() { sttService = stt.NewDeepgramService(cfg.AI.STT.APIKey, cfg.AI.STT.Model, cfg.AI.STT.Endpoint, cfg.AI.STT.Timeout, logger.Log) logger.Log.Infow("STT service initialized", "provider", "deepgram", "model", cfg.AI.STT.Model) } - llmService := llm.NewOpenAIService(cfg.AI.LLM.APIKey, cfg.AI.LLM.Model, cfg.AI.LLM.Endpoint, cfg.AI.LLM.Timeout, cfg.AI.LLM.HTTPClientTimeout, logger.Log) - logger.Log.Infow("LLM service initialized", "provider", cfg.AI.LLM.Provider, "model", cfg.AI.LLM.Model, "endpoint", cfg.AI.LLM.Endpoint, "timeout", cfg.AI.LLM.Timeout) - var ttsService tts.Service switch strings.ToLower(cfg.AI.TTS.Provider) { case "mimo", "xiaomi": @@ -129,8 +125,12 @@ func main() { logger.Log.Infow("TTS service initialized", "provider", "openai", "model", cfg.AI.TTS.Model, "voice", cfg.AI.TTS.Voice, "speed", cfg.AI.TTS.Speed) } - // 初始化 Orchestrator - orch := orchestrator.New(sttService, llmService, ttsService, sessionMgr, cfg) + // 初始化 Eino Graph + Orchestrator + pipelineGraph, err := eino.NewPipelineGraph(ctx, cfg, sttService, ttsService, sessionMgr) + if err != nil { + logger.Log.Fatalw("failed to create eino pipeline graph", "error", err) + } + orch := eino.NewEinoOrchestrator(pipelineGraph, sessionMgr, cfg.AI.LLM.Model) // 初始化认证服务 tokenMgr := auth.NewTokenManager( diff --git a/backend/go.mod b/backend/go.mod index 6a23eff..6f652f0 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -3,6 +3,8 @@ module github.com/hhs/camtalk go 1.25.0 require ( + github.com/cloudwego/eino v0.9.9 + github.com/cloudwego/eino-ext/components/model/openai v0.1.13 github.com/gin-gonic/gin v1.10.0 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/uuid v1.6.0 @@ -24,8 +26,6 @@ require ( github.com/bytedance/sonic/loader v0.5.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect - github.com/cloudwego/eino v0.9.9 // indirect - github.com/cloudwego/eino-ext/components/model/openai v0.1.13 // indirect github.com/cloudwego/eino-ext/libs/acl/openai v0.1.17 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/dustin/go-humanize v1.0.1 // indirect diff --git a/backend/go.sum b/backend/go.sum index 5c289db..9d58137 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -13,6 +13,8 @@ github.com/bugsnag/bugsnag-go v1.4.0/go.mod h1:2oa8nejYd4cQ/b0hMIopN0lCRxU0bueqR github.com/bugsnag/panicwrap v1.2.0/go.mod h1:D/8v3kj0zr8ZAKg1AQ6crr+5VwKN5eIywRkfhyM/+dE= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= +github.com/bytedance/mockey v1.3.0 h1:ONLRdvhqmCfr9rTasUB8ZKCfvbdD2tohOg4u+4Q/ed0= +github.com/bytedance/mockey v1.3.0/go.mod h1:1BPHF9sol5R1ud/+0VEHGQq/+i2lN+GTsr3O2Q9IENY= github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k= github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE= @@ -49,6 +51,8 @@ github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= github.com/gin-gonic/gin v1.10.0 h1:nTuyha1TYqgedzytsKYqna+DfLos46nTv2ygFy86HFU= github.com/gin-gonic/gin v1.10.0/go.mod h1:4PMNQiOhvDRa013RKVbsiNwoyezlm2rm0uX/T7kzp5Y= +github.com/go-check/check v0.0.0-20180628173108-788fd7840127 h1:0gkP6mzaMqkmpcJYCFOLkIBwI7xFExG03bbkOkCvUPI= +github.com/go-check/check v0.0.0-20180628173108-788fd7840127/go.mod h1:9ES+weclKsC9YodN5RgxqK/VD9HM9JsCSh7rNhMZE98= github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= @@ -72,6 +76,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/goph/emperror v0.17.2 h1:yLapQcmEsO0ipe9p5TaN22djm3OFV/TfM/fcYP0/J18= github.com/goph/emperror v0.17.2/go.mod h1:+ZbQ+fUNO/6FNiUo0ujtMjhgad9Xa6fQL9KhH4LNHic= +github.com/gopherjs/gopherjs v1.17.2 h1:fQnZVsXk8uxXIStYb0N4bGk7jeyTalG/wsZjQ25dO0g= +github.com/gopherjs/gopherjs v1.17.2/go.mod h1:pRRIvn/QzFLrKfvEz3qUuEhtE/zLCWfreZ6J5gM2i+k= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= @@ -89,6 +95,8 @@ github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwA github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= +github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0/go.mod h1:1NbS8ALrpOvjt0rHPNLyCIeMtbizbir8U//inJ+zuB8= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= @@ -104,10 +112,14 @@ github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= +github.com/mattn/go-colorable v0.1.2 h1:/bC9yWikZXAL9uJdulbSfyVNIR3n3trXl+v8+1sx8mU= +github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/meguminnnnnnnnn/go-openai v0.1.2 h1:iXombGGjqjBrmE9WaSidUhhi3YQhf42QTHvHLMkgvCA= github.com/meguminnnnnnnnn/go-openai v0.1.2/go.mod h1:qs96ysDmxhE4BZoU45I43zcyfnaYxU3X+aRzLko/htY= +github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b h1:j7+1HpAFS1zy5+Q4qx1fWh90gTKwiN4QCGoY9TWyyO4= +github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -137,6 +149,10 @@ github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/slongfield/pyfmt v0.0.0-20220222012616-ea85ff4c361f h1:Z2cODYsUxQPofhpYRMQVwWz4yUVpHF+vPi+eUdruUYI= github.com/slongfield/pyfmt v0.0.0-20220222012616-ea85ff4c361f/go.mod h1:JqzWyvTuI2X4+9wOHmKSQCYxybB/8j6Ko43qVmXDuZg= +github.com/smarty/assertions v1.15.0 h1:cR//PqUBUiQRakZWqBiFFQ9wb8emQGDb0HeGdqGByCY= +github.com/smarty/assertions v1.15.0/go.mod h1:yABtdzeQs6l1brC900WlRNwj6ZR55d7B+E8C6HtKdec= +github.com/smartystreets/goconvey v1.8.1 h1:qGjIddxOk4grTu9JPOU31tVfq3cNdBlNa5sSznIX1xY= +github.com/smartystreets/goconvey v1.8.1/go.mod h1:+/u4qLyY6x1jReYOp7GOM2FSt8aP9CzCZL03bI28W60= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U= github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= @@ -170,6 +186,8 @@ github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65E github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/fJgbpc= github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw= +github.com/x-cray/logrus-prefixed-formatter v0.5.2 h1:00txxvfBM9muc0jiLIEAkAcIMJzfthRT6usrui8uGmg= +github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE= github.com/yargevad/filepathx v1.0.0 h1:SYcT+N3tYGi+NvazubCNlvgIPbzAk7i7y2dwg3I5FYc= github.com/yargevad/filepathx v1.0.0/go.mod h1:BprfX/gpYNJHJfc35GjRRpVcwWXS89gGulUIU5tK3tA= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= @@ -178,6 +196,8 @@ go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/mock v0.4.0 h1:VcM4ZOtdbR4f6VXfiOpwpVJDL6lCReaZ6mw31wqh7KU= +go.uber.org/mock v0.4.0/go.mod h1:a6FSlNadKUHUa9IP5Vyt1zh4fC7uAwxMutEAscFbkZc= go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo= @@ -203,6 +223,8 @@ golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/term v0.28.0 h1:/Ts8HFuMR2E6IP/jlo7QVLZHggjKQbhu/7H0LJFr3Gg= +golang.org/x/term v0.28.0/go.mod h1:Sw/lC2IAUZ92udQNf3WodGtn4k/XoLyZoh8v/8uiwek= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= diff --git a/backend/internal/eino/adapter.go b/backend/internal/eino/adapter.go new file mode 100644 index 0000000..75c1b2c --- /dev/null +++ b/backend/internal/eino/adapter.go @@ -0,0 +1,165 @@ +package eino + +import ( + "context" + "encoding/base64" + "io" + "time" + + "github.com/cloudwego/eino/compose" + + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" + "github.com/hhs/camtalk/internal/orchestrator" + "github.com/hhs/camtalk/internal/session" +) + +// ctxKeySessionID sessionID 的 context key。 +type ctxKeySessionID struct{} + +// WithSessionID 将 sessionID 注入 context。 +func WithSessionID(ctx context.Context, sessionID string) context.Context { + return context.WithValue(ctx, ctxKeySessionID{}, sessionID) +} + +// EinoOrchestrator 实现 orchestrator.Orchestrator 接口。 +// 将 Eino Graph 包装为现有接口，WS Handler 几乎不用改。 +type EinoOrchestrator struct { + graph *PipelineGraph + sessionMgr session.Manager + model string + callbacks compose.Option // 运行时 Callback option +} + +// NewEinoOrchestrator 创建 Eino 编排器适配器。 +func NewEinoOrchestrator(graph *PipelineGraph, sessionMgr session.Manager, model string) *EinoOrchestrator { + return &EinoOrchestrator{ + graph: graph, + sessionMgr: sessionMgr, + model: model, + callbacks: compose.WithCallbacks(BuildCallbackHandler()), + } +} + +// ProcessQuery 实现 orchestrator.Orchestrator 接口。 +func (e *EinoOrchestrator) ProcessQuery( + ctx context.Context, + sessionID string, + req models.WsQuery, + history []models.Message, + sender orchestrator.Sender, +) error { + log := logger.Log + startTime := time.Now() + + // 1. 设置活跃请求 + if err := e.sessionMgr.SetActiveRequest(ctx, sessionID, req.RequestID); err != nil { + log.Errorw("设置活跃请求失败", "error", err) + } + defer e.sessionMgr.ClearActiveRequest(ctx, sessionID) + + // 2. 获取会话配置 + sess, err := e.sessionMgr.Get(ctx, sessionID) + if err != nil { + log.Errorw("获取会话失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "SESSION_NOT_FOUND", + Message: "会话不存在", + }) + return err + } + + // 3. 解码音频和图片 + var audioData []byte + if req.Text == "" && req.Audio != "" { + audioData, err = base64.StdEncoding.DecodeString(req.Audio) + if err != nil { + log.Errorw("音频解码失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "INVALID_MESSAGE", + Message: "音频数据解码失败", + }) + return err + } + } + + var imageData []byte + if req.Image != "" { + imageData, err = base64.StdEncoding.DecodeString(req.Image) + if err != nil { + log.Errorw("图片解码失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "INVALID_MESSAGE", + Message: "图片数据解码失败", + }) + return err + } + } + + // 4. 构建 Graph 输入 + input := buildPipelineInput(req, sessionID, sess, audioData, imageData) + + // 5. 注入 context 值（供 Callback 和 Lambda 节点使用） + ctx = WithSender(ctx, sender) + ctx = WithRequestID(ctx, req.RequestID) + ctx = WithSessionID(ctx, sessionID) + ctx = WithStartTime(ctx, startTime) + ctx = WithPipelineState(ctx, genLocalState(ctx)) + + // 6. 追加用户消息到历史 + if req.Text != "" { + _ = e.sessionMgr.AppendMessage(ctx, sessionID, models.Message{ + Role: "user", + Content: req.Text, + }) + } + + // 7. 调用 Graph（Stream 模式 + 运行时 Callback） + streamReader, err := e.graph.Runnable.Stream(ctx, input, e.callbacks) + if err != nil { + log.Errorw("Graph Stream 启动失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "INTERNAL_ERROR", + Message: "编排器启动失败", + }) + return err + } + + // 8. 消费 StreamReader（触发整条链路执行，side effects 推送消息到客户端） + var output PipelineOutput + for { + o, err := streamReader.Recv() + if err != nil { + if err == io.EOF { + break + } + log.Errorw("Graph Stream 消费错误", "error", err) + break + } + output = o + } + + // 9. 追加助手消息到历史 + if output.FullResponse != "" { + _ = e.sessionMgr.AppendMessage(ctx, sessionID, models.Message{ + Role: "assistant", + Content: output.FullResponse, + }) + } + + latency := time.Since(startTime).Milliseconds() + log.Infow("Eino 编排完成", + "request_id", req.RequestID, + "latency_ms", latency, + "session_id", sessionID) + + return nil +} diff --git a/backend/internal/eino/graph.go b/backend/internal/eino/graph.go new file mode 100644 index 0000000..f8a2449 --- /dev/null +++ b/backend/internal/eino/graph.go @@ -0,0 +1,113 @@ +package eino + +import ( + "context" + "time" + + openaiImpl "github.com/cloudwego/eino-ext/components/model/openai" + "github.com/cloudwego/eino/compose" + + "github.com/hhs/camtalk/internal/ai/stt" + "github.com/hhs/camtalk/internal/ai/tts" + "github.com/hhs/camtalk/internal/config" + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" + "github.com/hhs/camtalk/internal/session" +) + +const ( + nodeSTT = "stt" + nodeHistory = "history" + nodeLLM = "llm" + nodeSplitter = "splitter" + nodeTTS = "tts" + nodeDone = "done" +) + +// PipelineGraph 封装编译后的 Eino Graph。 +type PipelineGraph struct { + Runnable compose.Runnable[PipelineInput, PipelineOutput] +} + +// NewPipelineGraph 构建 CamTalk AI 编排 Graph。 +// +// 拓扑：START → STT → History → ChatModel → Splitter → TTS → Done → END +// +// Graph 使用 Stream 模式调用，ChatModel 实现真正的 token 级流式输出。 +// LLM token 通过 Callback 的 OnEndWithStreamOutput 实时推送到客户端。 +func NewPipelineGraph( + ctx context.Context, + cfg *config.Config, + sttService stt.Service, + ttsService tts.Service, + sessionMgr session.Manager, +) (*PipelineGraph, error) { + log := logger.Log + + // 1. 创建 eino-ext ChatModel（对接 DashScope OpenAI 兼容接口） + chatModel, err := openaiImpl.NewChatModel(ctx, &openaiImpl.ChatModelConfig{ + APIKey: cfg.AI.LLM.APIKey, + Model: cfg.AI.LLM.Model, + BaseURL: cfg.AI.LLM.Endpoint, + Timeout: time.Duration(cfg.AI.LLM.Timeout) * time.Second, + }) + if err != nil { + return nil, err + } + log.Infow("Eino ChatModel 初始化成功", + "model", cfg.AI.LLM.Model, + "endpoint", cfg.AI.LLM.Endpoint) + + // 2. 构建 Graph（值类型，非指针） + g := compose.NewGraph[PipelineInput, PipelineOutput]( + compose.WithGenLocalState(genLocalState), + ) + + // 3. 添加节点 + maxHistory := cfg.Session.MaxHistory + + _ = g.AddLambdaNode(nodeSTT, NewSTTLambda(sttService)) + _ = g.AddLambdaNode(nodeHistory, NewHistoryLambda(sessionMgr.GetHistory, maxHistory)) + _ = g.AddChatModelNode(nodeLLM, chatModel) + _ = g.AddLambdaNode(nodeSplitter, NewSplitterLambda()) + _ = g.AddLambdaNode(nodeTTS, NewTTSLambda( + ttsService, + cfg.AI.TTS.Voice, + cfg.AI.TTS.Speed, + cfg.AI.TTS.OutputFormat, + cfg.AI.TTS.SampleRate, + )) + _ = g.AddLambdaNode(nodeDone, NewDoneLambda(cfg.AI.LLM.Model)) + + // 4. 连接边 + _ = g.AddEdge(compose.START, nodeSTT) + _ = g.AddEdge(nodeSTT, nodeHistory) + _ = g.AddEdge(nodeHistory, nodeLLM) + _ = g.AddEdge(nodeLLM, nodeSplitter) + _ = g.AddEdge(nodeSplitter, nodeTTS) + _ = g.AddEdge(nodeTTS, nodeDone) + _ = g.AddEdge(nodeDone, compose.END) + + // 5. 编译（回调在运行时通过 Stream option 传入） + runnable, err := g.Compile(ctx) + if err != nil { + return nil, err + } + + log.Infow("Eino Graph 编译成功", "nodes", 6) + return &PipelineGraph{Runnable: runnable}, nil +} + +// buildPipelineInput 从 WebSocket 请求和会话配置构建 Graph 输入。 +func buildPipelineInput(req models.WsQuery, sessionID string, sess *models.Session, audioData, imageData []byte) PipelineInput { + return PipelineInput{ + AudioData: audioData, + ImageData: imageData, + Text: req.Text, + SessionID: sessionID, + RequestID: req.RequestID, + Language: sess.Config.Language, + Scenario: sess.Config.Scenario, + TTSEnabled: sess.Config.TTSEnabled, + } +} diff --git a/backend/internal/eino/nodes_done.go b/backend/internal/eino/nodes_done.go index 90eb362..745c69b 100644 --- a/backend/internal/eino/nodes_done.go +++ b/backend/internal/eino/nodes_done.go @@ -8,49 +8,47 @@ import ( "github.com/hhs/camtalk/internal/logger" "github.com/hhs/camtalk/internal/models" - "github.com/hhs/camtalk/internal/session" ) +// ctxKeyStartTime 请求开始时间的 context key。 +type ctxKeyStartTime struct{} + +// WithStartTime 将请求开始时间注入 context。 +func WithStartTime(ctx context.Context, t time.Time) context.Context { + return context.WithValue(ctx, ctxKeyStartTime{}, t) +} + +// latencyFromCtx 从 context 获取开始时间并计算延迟（毫秒）。 +func latencyFromCtx(ctx context.Context) int64 { + if startTime, ok := ctx.Value(ctxKeyStartTime{}).(time.Time); ok { + return time.Since(startTime).Milliseconds() + } + return 0 +} + // NewDoneLambda 创建 Done Lambda 节点。 -// 输入: struct{}（TTS 完成信号）→ 输出: PipelineOutput +// 输入: struct{}（TTS 完成信号）→ 输出: *PipelineOutput // -// 从 PipelineState 读取完整回复和 token 用量，发送 llm_done 到客户端， -// 追加助手消息到会话历史，返回 PipelineOutput。 -func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda { - return compose.InvokableLambda(func(ctx context.Context, _ struct{}) (*PipelineOutput, error) { +// 从 PipelineState 读取完整回复和 token 用量，发送 llm_done 到客户端。 +// 历史消息追加由适配器负责（避免重复写入）。 +func NewDoneLambda(defaultModel string) *compose.Lambda { + return compose.InvokableLambda(func(ctx context.Context, _ struct{}) (PipelineOutput, error) { log := logger.Log sender := senderFromCtx(ctx) - requestID := requestIDFromCtx(ctx) state := stateFromCtx(ctx) if state == nil { - return &PipelineOutput{}, nil + return PipelineOutput{}, nil } state.mu.Lock() fullResponse := state.FullResponse.String() transcribedText := state.TranscribedText tokenUsage := state.TokenUsage - modelName := model + requestID := state.RequestID + modelName := defaultModel state.mu.Unlock() - // 追加助手消息到会话历史 - sessionID := "" - if state != nil { - // 从 context 获取 sessionID（由适配器注入） - if sid, ok := ctx.Value(ctxKeySessionID{}).(string); ok { - sessionID = sid - } - } - if sessionID != "" && sessionMgr != nil && fullResponse != "" { - if err := sessionMgr.AppendMessage(ctx, sessionID, models.Message{ - Role: "assistant", - Content: fullResponse, - }); err != nil { - log.Errorw("追加助手消息到历史失败", "error", err) - } - } - // 发送 llm_done if sender != nil && requestID != "" { done := models.WsLLMDone{ @@ -58,7 +56,7 @@ func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda { RequestID: requestID, FullText: fullResponse, Model: modelName, - LatencyMs: 0, // 由适配器计算 + LatencyMs: latencyFromCtx(ctx), } if tokenUsage != nil { done.TokensUsed = struct { @@ -78,9 +76,9 @@ func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda { log.Infow("查询处理完成", "request_id", requestID, - "text_length", len(fullResponse)) + "response_length", len(fullResponse)) - return &PipelineOutput{ + return PipelineOutput{ TranscribedText: transcribedText, FullResponse: fullResponse, Model: modelName, @@ -88,27 +86,3 @@ func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda { }, nil }) } - -// ctxKeySessionID sessionID 的 context key。 -type ctxKeySessionID struct{} - -// WithSessionID 将 sessionID 注入 context。 -func WithSessionID(ctx context.Context, sessionID string) context.Context { - return context.WithValue(ctx, ctxKeySessionID{}, sessionID) -} - -// latencyFromCtx 从 context 获取开始时间并计算延迟。 -func latencyFromCtx(ctx context.Context) int64 { - if startTime, ok := ctx.Value(ctxKeyStartTime{}).(time.Time); ok { - return time.Since(startTime).Milliseconds() - } - return 0 -} - -// ctxKeyStartTime 请求开始时间的 context key。 -type ctxKeyStartTime struct{} - -// WithStartTime 将请求开始时间注入 context。 -func WithStartTime(ctx context.Context, t time.Time) context.Context { - return context.WithValue(ctx, ctxKeyStartTime{}, t) -} diff --git a/backend/internal/eino/nodes_history.go b/backend/internal/eino/nodes_history.go index 0858f24..a558a6b 100644 --- a/backend/internal/eino/nodes_history.go +++ b/backend/internal/eino/nodes_history.go @@ -3,6 +3,7 @@ package eino import ( "context" "encoding/base64" + "github.com/cloudwego/eino/compose" "github.com/cloudwego/eino/schema" @@ -11,28 +12,33 @@ import ( "github.com/hhs/camtalk/internal/models" ) -// HistoryInput 历史组装节点的输入，包含 STT 输出和原始请求信息。 -type HistoryInput struct { - STTOutput *STTOutput - SessionID string - RequestID string - ImageData []byte - Scenario string - DetailLevel string -} - // NewHistoryLambda 创建历史组装 Lambda 节点。 -// 输入: HistoryInput → 输出: []*schema.Message +// 输入: *STTOutput → 输出: []*schema.Message // +// 从 PipelineState 读取请求元数据（SessionID、Scenario、ImageData 等）， // 构建系统提示词，组装历史消息和当前用户输入（含多模态图片）。 -func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string, maxHistory int) ([]models.Message, error), maxHistory int) *compose.Lambda { - return compose.InvokableLambda(func(ctx context.Context, input *HistoryInput) ([]*schema.Message, error) { +func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string, limit int) ([]models.Message, error), maxHistory int) *compose.Lambda { + return compose.InvokableLambda(func(ctx context.Context, sttOut STTOutput) ([]*schema.Message, error) { log := logger.Log - requestID := input.RequestID + + // 从 State 读取请求元数据 + state := stateFromCtx(ctx) + if state == nil { + return []*schema.Message{}, nil + } + + state.mu.Lock() + sessionID := state.SessionID + requestID := state.RequestID + imageData := state.ImageData + scenario := state.Scenario + detailLevel := state.DetailLevel + language := sttOut.Language + state.mu.Unlock() // 构建系统提示词 - scenarioPrompt := llm.GetScenarioPrompt(input.Scenario, input.STTOutput.Language) - systemPrompt := llm.BuildSystemPrompt(input.STTOutput.Language, input.DetailLevel, scenarioPrompt) + scenarioPrompt := llm.GetScenarioPrompt(scenario, language) + systemPrompt := llm.BuildSystemPrompt(language, detailLevel, scenarioPrompt) // 构建 system message（含图片） systemMsg := &schema.Message{ @@ -41,9 +47,9 @@ func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string, } // 如果有图片，添加到 system message 的多模态内容中 - if len(input.ImageData) > 0 { - base64Str := base64.StdEncoding.EncodeToString(input.ImageData) - mimeType := detectImageMimeType(input.ImageData) + if len(imageData) > 0 { + base64Str := base64.StdEncoding.EncodeToString(imageData) + mimeType := detectImageMimeType(imageData) systemMsg.UserInputMultiContent = []schema.MessageInputPart{ { Type: schema.ChatMessagePartTypeImageURL, @@ -61,8 +67,8 @@ func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string, messages := []*schema.Message{systemMsg} // 获取并追加历史消息 - if historyFetcher != nil && input.SessionID != "" { - history, err := historyFetcher(ctx, input.SessionID, maxHistory) + if historyFetcher != nil && sessionID != "" { + history, err := historyFetcher(ctx, sessionID, maxHistory) if err != nil { log.Warnw("获取历史消息失败，继续处理", "error", err, "request_id", requestID) } else { @@ -78,14 +84,14 @@ func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string, // 追加当前用户输入 messages = append(messages, &schema.Message{ Role: schema.User, - Content: input.STTOutput.Text, + Content: sttOut.Text, }) log.Infow("历史组装完成", "request_id", requestID, "message_count", len(messages), - "has_image", len(input.ImageData) > 0, - "scenario", input.Scenario) + "has_image", len(imageData) > 0, + "scenario", scenario) return messages, nil }) @@ -96,22 +102,17 @@ func detectImageMimeType(data []byte) string { if len(data) < 4 { return "image/jpeg" } - // JPEG: FF D8 FF if data[0] == 0xFF && data[1] == 0xD8 && data[2] == 0xFF { return "image/jpeg" } - // PNG: 89 50 4E 47 if data[0] == 0x89 && data[1] == 0x50 && data[2] == 0x4E && data[3] == 0x47 { return "image/png" } - // GIF: 47 49 46 38 if data[0] == 0x47 && data[1] == 0x49 && data[2] == 0x46 { return "image/gif" } - // WebP: 52 49 46 46 if data[0] == 0x52 && data[1] == 0x49 && data[2] == 0x46 && data[3] == 0x46 { return "image/webp" } - return "image/jpeg" // 默认 + return "image/jpeg" } - diff --git a/backend/internal/eino/nodes_stt.go b/backend/internal/eino/nodes_stt.go index 6659863..0f63a05 100644 --- a/backend/internal/eino/nodes_stt.go +++ b/backend/internal/eino/nodes_stt.go @@ -19,11 +19,24 @@ import ( // 语音模式：调用 sttService.Recognize() 进行语音识别。 // 识别结果通过 Sender 发送 stt_result 到客户端。 func NewSTTLambda(sttService stt.Service) *compose.Lambda { - return compose.InvokableLambda(func(ctx context.Context, input *PipelineInput) (*STTOutput, error) { + return compose.InvokableLambda(func(ctx context.Context, input PipelineInput) (STTOutput, error) { log := logger.Log sender := senderFromCtx(ctx) requestID := requestIDFromCtx(ctx) + // 将输入元数据写入 State，供下游节点（History、Done）读取 + if state := stateFromCtx(ctx); state != nil { + state.mu.Lock() + state.SessionID = input.SessionID + state.RequestID = input.RequestID + state.ImageData = input.ImageData + state.Scenario = input.Scenario + state.DetailLevel = "low" + state.Language = input.Language + state.TTSEnabled = input.TTSEnabled + state.mu.Unlock() + } + // 文本输入模式：跳过 STT if input.Text != "" { log.Infow("使用文本输入，跳过 STT", @@ -48,7 +61,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda { state.mu.Unlock() } - return &STTOutput{ + return STTOutput{ Text: input.Text, Language: input.Language, IsSkipped: true, @@ -57,7 +70,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda { // 语音模式：解码音频 if len(input.AudioData) == 0 { - return nil, fmt.Errorf("stt: no audio data provided") + return STTOutput{}, fmt.Errorf("stt: no audio data provided") } log.Infow("开始语音识别", @@ -79,7 +92,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda { Message: "语音识别失败: " + err.Error(), }) } - return nil, fmt.Errorf("stt: recognize: %w", err) + return STTOutput{}, fmt.Errorf("stt: recognize: %w", err) } // STT 返回空文本 @@ -109,7 +122,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda { state.mu.Unlock() } - return &STTOutput{ + return STTOutput{ Text: text, Language: input.Language, IsSkipped: false, diff --git a/backend/internal/eino/state.go b/backend/internal/eino/state.go index 613cd9b..aef96af 100644 --- a/backend/internal/eino/state.go +++ b/backend/internal/eino/state.go @@ -7,13 +7,22 @@ import ( ) // PipelineState Graph 全局状态，用于跨节点收集数据。 -// 通过 compose.WithGenLocalState 注册，各节点通过 StatePreHandler/StatePostHandler 读写。 +// 通过 compose.WithGenLocalState 注册，各节点通过 compose.ProcessState 读写。 type PipelineState struct { mu sync.Mutex FullResponse strings.Builder // LLM 完整回复（由 Callback 累积） TranscribedText string // STT 识别文本 Model string // 实际使用的模型名 TokenUsage *TokenUsage // token 用量 + + // 从 PipelineInput 复制的元数据，供下游节点（History、Done）读取 + SessionID string + RequestID string + ImageData []byte + Scenario string + DetailLevel string + Language string + TTSEnabled bool } // genLocalState 创建每请求的 PipelineState 实例。