feat(api): 实现HTTP服务骨架与Server-Sent Events流式交互接口
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WeiCHEN
committed by
weichen
weichen
co-authored by
Cursor
Cursor
parent
0f01f047d5
commit
810eb26be3
@@ -0,0 +1,78 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/bizdev"
|
||||
"chat-agent-system/internal/config"
|
||||
"chat-agent-system/internal/server"
|
||||
)
|
||||
|
||||
func main() {
|
||||
envPath := flag.String("env", ".env", "环境配置文件路径")
|
||||
port := flag.Int("port", 0, "HTTP 服务端口 (默认读取配置文件或 8080)")
|
||||
host := flag.String("host", "", "HTTP 监听主机 (默认读取配置文件或 127.0.0.1)")
|
||||
backend := flag.String("backend", "", "智能体后端调度覆盖 (stub, pi-agent, cursor-cli)")
|
||||
flag.Parse()
|
||||
|
||||
cfg, err := config.Load(*envPath)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "加载配置失败: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
if *port > 0 {
|
||||
cfg.HTTPPort = *port
|
||||
}
|
||||
if *host != "" {
|
||||
cfg.HTTPHost = *host
|
||||
}
|
||||
if *backend != "" {
|
||||
cfg.AgentBackend = *backend
|
||||
}
|
||||
|
||||
svc := bizdev.NewService(cfg)
|
||||
srv := server.NewServer(cfg, svc)
|
||||
|
||||
addr := fmt.Sprintf("%s:%d", cfg.HTTPHost, cfg.HTTPPort)
|
||||
fmt.Println("==================================================================")
|
||||
fmt.Println(" 🤖 医疗器械商务与研发智能参谋系统 - HTTP / SSE API 守护服务 ")
|
||||
fmt.Println("==================================================================")
|
||||
fmt.Printf("服务环境: %s | 智能体后端: %s\n", cfg.Env, cfg.AgentBackend)
|
||||
fmt.Printf("监听地址: http://%s\n", addr)
|
||||
fmt.Println("核心服务路由:")
|
||||
fmt.Println(" • [GET] /api/v1/health - 服务健康检查与基础设施状态")
|
||||
fmt.Println(" • [GET] /api/v1/projects - 查询项目当前上下文与黑板事实")
|
||||
fmt.Println(" • [POST] /api/v1/chat - 标准阻塞式对话协同接口")
|
||||
fmt.Println(" • [POST] /api/v1/chat/stream - 核心 SSE 流式长连接对话协议 (分层按需)")
|
||||
fmt.Println(" • [GET] /api/v1/chat/stream - SSE 流式长连接对话调试接口 (EventSource)")
|
||||
fmt.Println(" • [POST] /api/v1/tasks/run/stream- 独立任务与阶段执行 SSE 监控流")
|
||||
fmt.Println("==================================================================")
|
||||
|
||||
stop := make(chan os.Signal, 1)
|
||||
signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
|
||||
|
||||
go func() {
|
||||
if err := srv.Start(); err != nil && err != http.ErrServerClosed {
|
||||
fmt.Fprintf(os.Stderr, "服务启动失败: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}()
|
||||
|
||||
<-stop
|
||||
fmt.Println("\n正在优雅停止服务...")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := srv.Shutdown(ctx); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "服务停止异常: %v\n", err)
|
||||
} else {
|
||||
fmt.Println("服务已安全退出。")
|
||||
}
|
||||
}
|
||||
@@ -11,6 +11,8 @@ import (
|
||||
type Config struct {
|
||||
Env string `json:"env"`
|
||||
DataDir string `json:"data_dir"`
|
||||
HTTPHost string `json:"http_host"`
|
||||
HTTPPort int `json:"http_port"`
|
||||
AgentBackend string `json:"agent_backend"`
|
||||
AgentTimeout time.Duration `json:"agent_timeout"`
|
||||
CursorAPIKey string `json:"cursor_api_key"`
|
||||
@@ -45,6 +47,8 @@ func Load(path string) (*Config, error) {
|
||||
cfg := &Config{
|
||||
Env: "dev",
|
||||
DataDir: "./data/dev",
|
||||
HTTPHost: "127.0.0.1",
|
||||
HTTPPort: 8080,
|
||||
AgentBackend: "pi-agent",
|
||||
AgentTimeout: 30 * time.Minute,
|
||||
CursorModel: "auto",
|
||||
@@ -102,6 +106,12 @@ func Load(path string) (*Config, error) {
|
||||
cfg.Env = val
|
||||
case "DATA_DIR":
|
||||
cfg.DataDir = val
|
||||
case "APP_PUBLISH_HOST", "HTTP_HOST":
|
||||
cfg.HTTPHost = val
|
||||
case "HTTP_PORT":
|
||||
if p, err := strconv.Atoi(val); err == nil {
|
||||
cfg.HTTPPort = p
|
||||
}
|
||||
case "AGENT_BACKEND":
|
||||
cfg.AgentBackend = val
|
||||
case "AGENT_TIMEOUT":
|
||||
@@ -165,6 +175,16 @@ func Load(path string) (*Config, error) {
|
||||
if v := os.Getenv("DATA_DIR"); v != "" {
|
||||
cfg.DataDir = v
|
||||
}
|
||||
if v := os.Getenv("HTTP_HOST"); v != "" {
|
||||
cfg.HTTPHost = v
|
||||
} else if v := os.Getenv("APP_PUBLISH_HOST"); v != "" {
|
||||
cfg.HTTPHost = v
|
||||
}
|
||||
if v := os.Getenv("HTTP_PORT"); v != "" {
|
||||
if p, err := strconv.Atoi(v); err == nil {
|
||||
cfg.HTTPPort = p
|
||||
}
|
||||
}
|
||||
if v := os.Getenv("AGENT_BACKEND"); v != "" {
|
||||
cfg.AgentBackend = v
|
||||
}
|
||||
|
||||
@@ -0,0 +1,170 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
|
||||
"chat-agent-system/internal/bizdev"
|
||||
)
|
||||
|
||||
// HealthResponse 健康检查响应
|
||||
type HealthResponse struct {
|
||||
Status string `json:"status"`
|
||||
Env string `json:"env"`
|
||||
AgentBackend string `json:"agent_backend"`
|
||||
Version string `json:"version"`
|
||||
SharedKBPort int `json:"shared_kb_port"`
|
||||
}
|
||||
|
||||
// ChatRequest 对话请求
|
||||
type ChatRequest struct {
|
||||
SessionID string `json:"session_id"`
|
||||
Message string `json:"message"`
|
||||
ProjectName string `json:"project_name,omitempty"`
|
||||
CustomerName string `json:"customer_name,omitempty"`
|
||||
DeviceType string `json:"device_type,omitempty"`
|
||||
IntendedUse string `json:"intended_use,omitempty"`
|
||||
TargetMarket string `json:"target_market,omitempty"`
|
||||
WorkDir string `json:"work_dir,omitempty"`
|
||||
UserName string `json:"user_name,omitempty"`
|
||||
UserRole string `json:"user_role,omitempty"`
|
||||
}
|
||||
|
||||
// TaskRunRequestDTO 独立任务运行请求
|
||||
type TaskRunRequestDTO struct {
|
||||
TaskKey string `json:"task_key"`
|
||||
Stage string `json:"stage,omitempty"` // 若指定 stage，则执行成熟度阶段跃迁
|
||||
ProjectName string `json:"project_name"`
|
||||
CustomerName string `json:"customer_name"`
|
||||
DeviceType string `json:"device_type"`
|
||||
IntendedUse string `json:"intended_use"`
|
||||
TargetMarket string `json:"target_market"`
|
||||
Fields map[string]string `json:"fields,omitempty"`
|
||||
Backend string `json:"backend,omitempty"`
|
||||
AutoRepair bool `json:"auto_repair"`
|
||||
}
|
||||
|
||||
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
|
||||
resp := HealthResponse{
|
||||
Status: "ok",
|
||||
Env: s.cfg.Env,
|
||||
AgentBackend: s.cfg.AgentBackend,
|
||||
Version: "v1.0",
|
||||
SharedKBPort: s.cfg.KbDbPort,
|
||||
}
|
||||
writeJSONResponse(w, http.StatusOK, resp)
|
||||
}
|
||||
|
||||
func (s *Server) handleGetProject(w http.ResponseWriter, r *http.Request) {
|
||||
sessionID := r.URL.Query().Get("session_id")
|
||||
if sessionID == "" {
|
||||
http.Error(w, "missing session_id query param", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
s.sessionsMu.RLock()
|
||||
sess, exists := s.sessions[sessionID]
|
||||
s.sessionsMu.RUnlock()
|
||||
|
||||
if !exists || sess == nil || sess.UserCtx == nil {
|
||||
http.Error(w, "session or project context not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
writeJSONResponse(w, http.StatusOK, sess.UserCtx)
|
||||
}
|
||||
|
||||
func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
|
||||
var req ChatRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if req.Message == "" {
|
||||
http.Error(w, "message cannot be empty", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
sess := s.getOrCreateSession(
|
||||
req.SessionID,
|
||||
req.ProjectName,
|
||||
req.CustomerName,
|
||||
req.DeviceType,
|
||||
req.IntendedUse,
|
||||
req.TargetMarket,
|
||||
req.WorkDir,
|
||||
req.UserName,
|
||||
req.UserRole,
|
||||
)
|
||||
|
||||
resp, err := sess.SendMessage(r.Context(), req.Message)
|
||||
if err != nil {
|
||||
http.Error(w, "process chat message failed: "+err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
writeJSONResponse(w, http.StatusOK, resp)
|
||||
}
|
||||
|
||||
func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) {
|
||||
var req TaskRunRequestDTO
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// 1. 如果指定了 stage，则执行成熟度阶段编排
|
||||
if req.Stage != "" {
|
||||
stageRes, err := s.svc.ExecuteStage(r.Context(), bizdev.StageRunRequest{
|
||||
Stage: bizdev.MaturityStage(req.Stage),
|
||||
ProjectName: req.ProjectName,
|
||||
CustomerName: req.CustomerName,
|
||||
DeviceType: req.DeviceType,
|
||||
IntendedUse: req.IntendedUse,
|
||||
TargetMarket: req.TargetMarket,
|
||||
Fields: req.Fields,
|
||||
Backend: req.Backend,
|
||||
AutoRepair: req.AutoRepair,
|
||||
})
|
||||
if err != nil {
|
||||
http.Error(w, "execute stage failed: "+err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
writeJSONResponse(w, http.StatusOK, stageRes)
|
||||
return
|
||||
}
|
||||
|
||||
// 2. 执行单任务
|
||||
if req.TaskKey == "" {
|
||||
req.TaskKey = "intake"
|
||||
}
|
||||
taskRes, err := s.svc.ExecuteTask(r.Context(), bizdev.TaskRunRequest{
|
||||
TaskKey: req.TaskKey,
|
||||
ProjectName: req.ProjectName,
|
||||
CustomerName: req.CustomerName,
|
||||
DeviceType: req.DeviceType,
|
||||
IntendedUse: req.IntendedUse,
|
||||
TargetMarket: req.TargetMarket,
|
||||
Fields: req.Fields,
|
||||
Backend: req.Backend,
|
||||
AutoRepair: req.AutoRepair,
|
||||
})
|
||||
if err != nil {
|
||||
http.Error(w, "execute task failed: "+err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// 隐藏不必要的中间大体积字节
|
||||
if taskRes.WorkDir != "" {
|
||||
taskRes.WorkDir = filepath.Clean(taskRes.WorkDir)
|
||||
}
|
||||
|
||||
writeJSONResponse(w, http.StatusOK, taskRes)
|
||||
}
|
||||
|
||||
func writeJSONResponse(w http.ResponseWriter, status int, data any) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(data)
|
||||
}
|
||||
@@ -0,0 +1,194 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/bizdev"
|
||||
"chat-agent-system/internal/chat"
|
||||
"chat-agent-system/internal/config"
|
||||
"chat-agent-system/internal/coordinator"
|
||||
"chat-agent-system/internal/intent"
|
||||
)
|
||||
|
||||
// Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务
|
||||
type Server struct {
|
||||
cfg *config.Config
|
||||
svc *bizdev.Service
|
||||
rec intent.Recognizer
|
||||
engine *coordinator.CoordinatorEngine
|
||||
sessions map[string]*chat.Session
|
||||
sessionsMu sync.RWMutex
|
||||
httpServer *http.Server
|
||||
}
|
||||
|
||||
// NewServer 构造函数
|
||||
func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
|
||||
if cfg == nil {
|
||||
cfg, _ = config.Load("")
|
||||
}
|
||||
if svc == nil {
|
||||
svc = bizdev.NewService(cfg)
|
||||
}
|
||||
|
||||
rec := intent.NewRecognizer()
|
||||
coordEngine := coordinator.NewCoordinatorEngine(svc)
|
||||
|
||||
s := &Server{
|
||||
cfg: cfg,
|
||||
svc: svc,
|
||||
rec: rec,
|
||||
engine: coordEngine,
|
||||
sessions: make(map[string]*chat.Session),
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// Routes 构建 HTTP 路由分发器
|
||||
func (s *Server) Routes() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
|
||||
// 1. 健康检查与系统自检
|
||||
mux.HandleFunc("GET /health", s.handleHealth)
|
||||
mux.HandleFunc("GET /api/v1/health", s.handleHealth)
|
||||
|
||||
// 2. 项目状态与会话查询
|
||||
mux.HandleFunc("GET /api/v1/projects", s.handleGetProject)
|
||||
|
||||
// 3. 对话协同服务 (标准 JSON 阻塞)
|
||||
mux.HandleFunc("POST /api/v1/chat", s.handleChat)
|
||||
|
||||
// 4. 对话协同服务 (核心 SSE 流式长连接)
|
||||
mux.HandleFunc("POST /api/v1/chat/stream", s.handleChatStream)
|
||||
mux.HandleFunc("GET /api/v1/chat/stream", s.handleChatStream)
|
||||
|
||||
// 5. 独立任务 / 阶段执行
|
||||
mux.HandleFunc("POST /api/v1/tasks/run", s.handleRunTask)
|
||||
mux.HandleFunc("POST /api/v1/tasks/run/stream", s.handleRunTaskStream)
|
||||
mux.HandleFunc("GET /api/v1/tasks/run/stream", s.handleRunTaskStream)
|
||||
|
||||
return s.corsMiddleware(mux)
|
||||
}
|
||||
|
||||
// Start 启动 HTTP 守护进程
|
||||
func (s *Server) Start() error {
|
||||
addr := fmt.Sprintf("%s:%d", s.cfg.HTTPHost, s.cfg.HTTPPort)
|
||||
s.httpServer = &http.Server{
|
||||
Addr: addr,
|
||||
Handler: s.Routes(),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
}
|
||||
|
||||
return s.httpServer.ListenAndServe()
|
||||
}
|
||||
|
||||
// Shutdown 优雅停机
|
||||
func (s *Server) Shutdown(ctx context.Context) error {
|
||||
if s.httpServer != nil {
|
||||
return s.httpServer.Shutdown(ctx)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// getOrCreateSession 获取或动态构建会话
|
||||
func (s *Server) getOrCreateSession(
|
||||
sessionID string,
|
||||
projectName string,
|
||||
customerName string,
|
||||
deviceType string,
|
||||
intendedUse string,
|
||||
targetMarket string,
|
||||
workDir string,
|
||||
userName string,
|
||||
userRole string,
|
||||
) *chat.Session {
|
||||
if sessionID == "" {
|
||||
sessionID = fmt.Sprintf("sess_%d", time.Now().UnixNano())
|
||||
}
|
||||
|
||||
s.sessionsMu.RLock()
|
||||
existing, exists := s.sessions[sessionID]
|
||||
s.sessionsMu.RUnlock()
|
||||
if exists && existing != nil {
|
||||
return existing
|
||||
}
|
||||
|
||||
s.sessionsMu.Lock()
|
||||
defer s.sessionsMu.Unlock()
|
||||
// 双重检查
|
||||
if existing, exists := s.sessions[sessionID]; exists && existing != nil {
|
||||
return existing
|
||||
}
|
||||
|
||||
if userName == "" {
|
||||
userName = "用户"
|
||||
}
|
||||
if userRole == "" {
|
||||
userRole = "engineer"
|
||||
}
|
||||
|
||||
u := coordinator.UserInfo{
|
||||
ID: "u_" + sessionID,
|
||||
Name: userName,
|
||||
Role: userRole,
|
||||
ProjectRole: "composer",
|
||||
}
|
||||
|
||||
var userCtx *coordinator.UserProjectContext
|
||||
if workDir != "" {
|
||||
assembler := coordinator.NewContextAssembler()
|
||||
userCtx, _ = assembler.AssembleFromWorkspace(workDir, u)
|
||||
}
|
||||
|
||||
if userCtx == nil {
|
||||
if projectName == "" {
|
||||
projectName = "医疗器械研发项目"
|
||||
}
|
||||
if customerName == "" {
|
||||
customerName = "水木东方客户"
|
||||
}
|
||||
userCtx = &coordinator.UserProjectContext{
|
||||
SnapshotID: fmt.Sprintf("ctx_%d", time.Now().UnixNano()),
|
||||
AssembledAt: time.Now(),
|
||||
User: u,
|
||||
WorkDir: workDir,
|
||||
Project: coordinator.ProjectInfo{
|
||||
Name: projectName,
|
||||
CustomerName: customerName,
|
||||
DeviceType: deviceType,
|
||||
IntendedUse: intendedUse,
|
||||
TargetMarket: targetMarket,
|
||||
CurrentWorkflow: "intake",
|
||||
CurrentStage: bizdev.StageV01_Triage,
|
||||
},
|
||||
BlackboardFacts: make(map[string]string),
|
||||
AvailableActions: coordinator.ComputeAvailableActions(bizdev.StageV01_Triage, u.ProjectRole),
|
||||
}
|
||||
if targetMarket != "" {
|
||||
userCtx.BlackboardFacts["target_market"] = targetMarket
|
||||
}
|
||||
if deviceType != "" {
|
||||
userCtx.BlackboardFacts["device_type"] = deviceType
|
||||
}
|
||||
}
|
||||
|
||||
sess := chat.NewSession(sessionID, userCtx, s.rec, s.engine)
|
||||
s.sessions[sessionID] = sess
|
||||
return sess
|
||||
}
|
||||
|
||||
func (s *Server) corsMiddleware(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Access-Control-Allow-Origin", "*")
|
||||
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS, PUT, DELETE")
|
||||
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, Accept")
|
||||
if r.Method == http.MethodOptions {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"chat-agent-system/internal/bizdev"
|
||||
"chat-agent-system/internal/chat"
|
||||
"chat-agent-system/internal/config"
|
||||
"chat-agent-system/internal/stream"
|
||||
)
|
||||
|
||||
func newTestServer() *Server {
|
||||
cfg := &config.Config{
|
||||
Env: "test",
|
||||
AgentBackend: "stub",
|
||||
DataDir: "./data/test",
|
||||
KbDbPort: 5432,
|
||||
}
|
||||
svc := bizdev.NewService(cfg)
|
||||
return NewServer(cfg, svc)
|
||||
}
|
||||
|
||||
func TestServer_Health(t *testing.T) {
|
||||
srv := newTestServer()
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/health", nil)
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
srv.Routes().ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected status 200, got %d", w.Code)
|
||||
}
|
||||
|
||||
var resp HealthResponse
|
||||
if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode response failed: %v", err)
|
||||
}
|
||||
|
||||
if resp.Status != "ok" || resp.Env != "test" {
|
||||
t.Errorf("unexpected health response: %+v", resp)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServer_ChatBlocking(t *testing.T) {
|
||||
srv := newTestServer()
|
||||
|
||||
chatReq := ChatRequest{
|
||||
SessionID: "test_sess_01",
|
||||
Message: "查一下当前进展",
|
||||
ProjectName: "内窥镜研发项目",
|
||||
CustomerName: "水木东方",
|
||||
}
|
||||
body, _ := json.Marshal(chatReq)
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/chat", bytes.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
srv.Routes().ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected status 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
var resp chat.ChatResponse
|
||||
if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode chat response failed: %v", err)
|
||||
}
|
||||
|
||||
if resp.ReplyText == "" {
|
||||
t.Errorf("expected non-empty reply text")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServer_ChatSSEStream(t *testing.T) {
|
||||
srv := newTestServer()
|
||||
|
||||
chatReq := StreamChatRequest{
|
||||
SessionID: "test_sess_sse",
|
||||
Message: "查一下当前进展",
|
||||
IncludeDetails: false,
|
||||
ProjectName: "内窥镜研发项目",
|
||||
CustomerName: "水木东方",
|
||||
}
|
||||
body, _ := json.Marshal(chatReq)
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/chat/stream", bytes.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
srv.Routes().ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected status 200, got %d", w.Code)
|
||||
}
|
||||
|
||||
contentType := w.Header().Get("Content-Type")
|
||||
if !strings.Contains(contentType, "text/event-stream") {
|
||||
t.Fatalf("expected Content-Type text/event-stream, got %s", contentType)
|
||||
}
|
||||
|
||||
// 解析 SSE 事件流
|
||||
scanner := bufio.NewScanner(w.Body)
|
||||
var currentEvent string
|
||||
var eventsReceived []string
|
||||
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if strings.HasPrefix(line, "event: ") {
|
||||
currentEvent = strings.TrimPrefix(line, "event: ")
|
||||
} else if strings.HasPrefix(line, "data: ") {
|
||||
dataStr := strings.TrimPrefix(line, "data: ")
|
||||
var ev stream.StreamEvent
|
||||
if err := json.Unmarshal([]byte(dataStr), &ev); err == nil {
|
||||
eventsReceived = append(eventsReceived, currentEvent)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(eventsReceived) == 0 {
|
||||
t.Fatalf("expected to receive SSE events, got 0")
|
||||
}
|
||||
|
||||
hasPhase := false
|
||||
hasStatus := false
|
||||
hasDone := false
|
||||
|
||||
for _, et := range eventsReceived {
|
||||
switch et {
|
||||
case "phase":
|
||||
hasPhase = true
|
||||
case "status":
|
||||
hasStatus = true
|
||||
case "done":
|
||||
hasDone = true
|
||||
}
|
||||
}
|
||||
|
||||
if !hasPhase {
|
||||
t.Errorf("expected phase event in SSE stream")
|
||||
}
|
||||
if !hasStatus {
|
||||
t.Errorf("expected status event in SSE stream")
|
||||
}
|
||||
if !hasDone {
|
||||
t.Errorf("expected done event in SSE stream")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,206 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/bizdev"
|
||||
"chat-agent-system/internal/stream"
|
||||
)
|
||||
|
||||
// StreamChatRequest SSE流式请求参数
|
||||
type StreamChatRequest struct {
|
||||
SessionID string `json:"session_id"`
|
||||
Message string `json:"message"`
|
||||
IncludeDetails bool `json:"include_details"` // 是否推送底层长篇推演草稿 (按需展开，默认 false)
|
||||
ProjectName string `json:"project_name,omitempty"`
|
||||
CustomerName string `json:"customer_name,omitempty"`
|
||||
DeviceType string `json:"device_type,omitempty"`
|
||||
IntendedUse string `json:"intended_use,omitempty"`
|
||||
TargetMarket string `json:"target_market,omitempty"`
|
||||
WorkDir string `json:"work_dir,omitempty"`
|
||||
UserName string `json:"user_name,omitempty"`
|
||||
UserRole string `json:"user_role,omitempty"`
|
||||
}
|
||||
|
||||
func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
|
||||
flusher, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
http.Error(w, "streaming unsupported by underlying connection", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
var req StreamChatRequest
|
||||
if r.Method == http.MethodPost {
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
} else {
|
||||
// 允许 GET query params 方便 EventSource 及浏览器直接调试
|
||||
q := r.URL.Query()
|
||||
req.SessionID = q.Get("session_id")
|
||||
req.Message = q.Get("message")
|
||||
req.ProjectName = q.Get("project_name")
|
||||
req.CustomerName = q.Get("customer_name")
|
||||
req.DeviceType = q.Get("device_type")
|
||||
req.IntendedUse = q.Get("intended_use")
|
||||
req.TargetMarket = q.Get("target_market")
|
||||
req.WorkDir = q.Get("work_dir")
|
||||
req.UserName = q.Get("user_name")
|
||||
req.UserRole = q.Get("user_role")
|
||||
if inc := q.Get("include_details"); inc != "" {
|
||||
req.IncludeDetails, _ = strconv.ParseBool(inc)
|
||||
}
|
||||
}
|
||||
|
||||
if req.Message == "" {
|
||||
http.Error(w, "message cannot be empty", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// 1. 设置 SSE 响应头
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
w.Header().Set("Access-Control-Allow-Origin", "*")
|
||||
w.Header().Set("X-Accel-Buffering", "no") // 禁用 Nginx 等反代缓冲
|
||||
flusher.Flush()
|
||||
|
||||
// 2. 准备会话
|
||||
sess := s.getOrCreateSession(
|
||||
req.SessionID,
|
||||
req.ProjectName,
|
||||
req.CustomerName,
|
||||
req.DeviceType,
|
||||
req.IntendedUse,
|
||||
req.TargetMarket,
|
||||
req.WorkDir,
|
||||
req.UserName,
|
||||
req.UserRole,
|
||||
)
|
||||
|
||||
// 3. 构建线程安全的 SSE 监听分发闭包
|
||||
var writeMu sync.Mutex
|
||||
listener := func(ev stream.StreamEvent) {
|
||||
writeMu.Lock()
|
||||
defer writeMu.Unlock()
|
||||
|
||||
data, err := json.Marshal(ev)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
// 按照标准 SSE 协议规范发射: event: <type>\ndata: <json>\n\n
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data))
|
||||
flusher.Flush()
|
||||
}
|
||||
|
||||
// 4. 发起流式执行
|
||||
_, err := sess.SendMessageStream(r.Context(), req.Message, listener, req.IncludeDetails)
|
||||
if err != nil {
|
||||
writeMu.Lock()
|
||||
errEv := stream.StreamEvent{
|
||||
Type: stream.EventError,
|
||||
Timestamp: time.Now(),
|
||||
Error: err.Error(),
|
||||
}
|
||||
data, _ := json.Marshal(errEv)
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", errEv.Type, string(data))
|
||||
flusher.Flush()
|
||||
writeMu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) {
|
||||
flusher, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
var req TaskRunRequestDTO
|
||||
includeDetails := false
|
||||
if r.Method == http.MethodPost {
|
||||
_ = json.NewDecoder(r.Body).Decode(&req)
|
||||
} else {
|
||||
q := r.URL.Query()
|
||||
req.TaskKey = q.Get("task_key")
|
||||
req.Stage = q.Get("stage")
|
||||
req.ProjectName = q.Get("project_name")
|
||||
req.CustomerName = q.Get("customer_name")
|
||||
req.DeviceType = q.Get("device_type")
|
||||
req.IntendedUse = q.Get("intended_use")
|
||||
req.TargetMarket = q.Get("target_market")
|
||||
req.Backend = q.Get("backend")
|
||||
req.AutoRepair, _ = strconv.ParseBool(q.Get("auto_repair"))
|
||||
if inc := q.Get("include_details"); inc != "" {
|
||||
includeDetails, _ = strconv.ParseBool(inc)
|
||||
}
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
w.Header().Set("Access-Control-Allow-Origin", "*")
|
||||
flusher.Flush()
|
||||
|
||||
var writeMu sync.Mutex
|
||||
listener := func(ev stream.StreamEvent) {
|
||||
writeMu.Lock()
|
||||
defer writeMu.Unlock()
|
||||
data, err := json.Marshal(ev)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data))
|
||||
flusher.Flush()
|
||||
}
|
||||
|
||||
ctxWithS, streamer := stream.WithListener(r.Context(), includeDetails, listener)
|
||||
start := time.Now()
|
||||
|
||||
if req.Stage != "" {
|
||||
_, err := s.svc.ExecuteStage(ctxWithS, bizdev.StageRunRequest{
|
||||
Stage: bizdev.MaturityStage(req.Stage),
|
||||
ProjectName: req.ProjectName,
|
||||
CustomerName: req.CustomerName,
|
||||
DeviceType: req.DeviceType,
|
||||
IntendedUse: req.IntendedUse,
|
||||
TargetMarket: req.TargetMarket,
|
||||
Fields: req.Fields,
|
||||
Backend: req.Backend,
|
||||
AutoRepair: req.AutoRepair,
|
||||
})
|
||||
if err != nil {
|
||||
streamer.EmitError(err)
|
||||
} else {
|
||||
streamer.EmitDone("stage_run", time.Since(start))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if req.TaskKey == "" {
|
||||
req.TaskKey = "intake"
|
||||
}
|
||||
|
||||
_, err := s.svc.ExecuteTask(ctxWithS, bizdev.TaskRunRequest{
|
||||
TaskKey: req.TaskKey,
|
||||
ProjectName: req.ProjectName,
|
||||
CustomerName: req.CustomerName,
|
||||
DeviceType: req.DeviceType,
|
||||
IntendedUse: req.IntendedUse,
|
||||
TargetMarket: req.TargetMarket,
|
||||
Fields: req.Fields,
|
||||
Backend: req.Backend,
|
||||
AutoRepair: req.AutoRepair,
|
||||
})
|
||||
if err != nil {
|
||||
streamer.EmitError(err)
|
||||
} else {
|
||||
streamer.EmitDone("task_run", time.Since(start))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user