Files
Cyrene/backend/ai-core/cmd/main.go
T
AskaEth 26a61cb57c feat: 第四轮大版本更新 — 修复4个严重Bug、2个UI Bug,实现自主思考重构与主-子会话架构
## 🐛 Bug 修复
- 修复前端对话无响应:消除 ChatContainer 中的双重 WebSocket 连接,优化 sendMessage 失败提示
- 修复 Memory-Service 数据库迁移失败:ai-core 和 memory-service 均添加 ALTER TABLE ADD COLUMN IF NOT EXISTS 模式演化
- 修复语音/STT 不可用:添加 MediaRecorder API 降级方案,修复 whisper-cli 输出文件名错误
- 修复仪表盘数据库按钮失效:补充按钮 ID 属性,重写 controlDB() 控制逻辑

## 🎨 UI 修复
- 修正用户消息头像位置:从 flex-row-reverse 改为 justify-end
- 移除空聊天列表的 emoji 占位图标

##  新功能
- devtools 新增 STT 处理日志面板(环形缓冲区 + WebSocket 广播 + 可视化表格)
- 新增 ADMIN_NICKNAME 环境变量,支持自定义管理员昵称

## 🔧 改进
- 注册流程增加昵称必填字段(前后端同步)

## 🏗️ 架构重构
- 重构自主思考逻辑:从定时器轮询改为事件驱动(对话后触发 + 静默检测),优化提示词使其更自然人性化
- 实现主-子会话架构:新增 4 种子会话类型(general/memory/iot/knowledge),意图分析 → 并行分发 → 结果合成流程

## 📄 新增文档
- docs/architecture/main-session-sub-session-design.md — 子会话架构设计文档
2026-05-19 21:09:48 +08:00

670 lines
18 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"os/signal"
"strings"
"syscall"
"time"
"github.com/joho/godotenv"
"github.com/yourname/cyrene-ai/ai-core/internal/background"
ctxbuild "github.com/yourname/cyrene-ai/ai-core/internal/context"
"github.com/yourname/cyrene-ai/ai-core/internal/llm"
"github.com/yourname/cyrene-ai/ai-core/internal/memory"
"github.com/yourname/cyrene-ai/ai-core/internal/model"
"github.com/yourname/cyrene-ai/ai-core/internal/orchestrator"
"github.com/yourname/cyrene-ai/ai-core/internal/persona"
"github.com/yourname/cyrene-ai/ai-core/internal/subsession"
"github.com/yourname/cyrene-ai/ai-core/internal/tools"
)
var cfg Config
func main() {
// 自动加载 .env 文件(来自 backend/.env
if err := godotenv.Load("../.env"); err != nil {
log.Println("ℹ 未找到 .env 文件,将使用环境变量或默认值")
}
log.SetFlags(log.LstdFlags | log.Lshortfile)
log.Println("🧠 AI-Core 服务启动中...")
// 加载配置
cfg = loadConfig()
// 初始化人格加载器
personaDir := cfg.PersonaDir
if personaDir == "" {
personaDir = "./internal/persona"
}
personaLoader, err := persona.NewLoader(personaDir)
if err != nil {
log.Fatalf("加载人格配置失败: %v", err)
}
log.Printf("已加载 %d 个人格: %v", len(personaLoader.List()), personaLoader.List())
// 初始化LLM适配器
llmProvider := llm.NewOpenAIProvider(llm.OpenAIConfig{
BaseURL: cfg.LLMBaseURL,
APIKey: cfg.LLMAPIKey,
Model: cfg.LLMModel,
FallbackModel: cfg.LLMFallbackModel,
Timeout: 120 * time.Second,
})
llmAdapter := llm.NewAdapter(llmProvider)
log.Printf("LLM适配器已就绪: 模型=%s", llmAdapter.ModelName())
// 初始化记忆系统
var memStore *memory.Store
var memRetriever *memory.Retriever
var memExtractor *memory.Extractor
if cfg.DatabaseURL != "" {
memStore = memory.NewStore(cfg.DatabaseURL)
defer memStore.Close()
memRetriever = memory.NewRetriever(memStore, nil)
// 记忆提取器使用LLM
memExtractor = memory.NewExtractor(memStore, func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) {
return llmAdapter.Chat(ctx, messages)
})
log.Println("记忆提取器已就绪")
}
// 初始化会话历史存储
convStore := ctxbuild.NewConversationStore(50)
log.Println("会话历史存储已就绪 (上限50条)")
// 初始化上下文构建器
ctxBuilder := ctxbuild.NewBuilder(convStore)
// 初始化 IoT 客户端
var iotClient *tools.IoTClient
if cfg.IoTServiceURL != "" {
iotClient = tools.NewIoTClient(cfg.IoTServiceURL)
log.Printf("IoT 客户端已就绪: %s", cfg.IoTServiceURL)
} else {
log.Println("IoT 客户端未配置 (IOT_DEBUG_SERVICE_URL 为空)")
}
// 初始化工具注册中心
toolRegistry := tools.NewRegistry()
if getEnvBool("ENABLE_TOOLS", true) {
toolRegistry.Register(tools.NewWebFetchTool())
toolRegistry.Register(tools.NewWebSearchTool())
toolRegistry.Register(tools.NewCalculatorTool())
toolRegistry.Register(tools.NewDateTimeTool())
toolRegistry.Register(tools.NewHTTPTool())
toolRegistry.Register(tools.NewJSONTool())
toolRegistry.Register(tools.NewTextTool())
toolRegistry.Register(tools.NewRandomTool())
toolRegistry.Register(tools.NewCryptoTool())
toolRegistry.Register(tools.NewMarkdownTool())
// File tool uses DATA_DIR or defaults to /tmp/cyrene_data
dataDir := getEnv("DATA_DIR", "/tmp/cyrene_data")
toolRegistry.Register(tools.NewFileTool(dataDir))
if iotClient != nil {
toolRegistry.Register(tools.NewIoTQueryTool(iotClient))
toolRegistry.Register(tools.NewIoTControlTool(iotClient))
}
log.Printf("工具注册中心已就绪: %d 个工具 (%v)", len(toolRegistry.ListTools()), toolRegistry.ListTools())
}
// 初始化后台思考器(增强版:支持工具调用和记忆管理)
thinkerCfg := background.DefaultThinkerConfig()
adminUserID := "admin_admin"
adminSessionID := "admin-session-main"
// 创建记忆服务 HTTP 客户端(用于持久化思考日志到 memory-service
memServiceURL := getEnv("MEMORY_SERVICE_URL", "http://localhost:8091")
memClient := memory.NewClient(memServiceURL)
log.Printf("记忆服务客户端已就绪: %s", memServiceURL)
thinker := background.NewThinker(
thinkerCfg,
personaLoader,
memRetriever,
llmAdapter,
iotClient,
memStore,
memExtractor,
toolRegistry,
convStore,
adminUserID,
adminSessionID,
memClient,
)
thinker.Start()
defer thinker.Stop()
// 健康检查与对话API的HTTP mux
mux := http.NewServeMux()
// 初始化子会话管理器
subManager := subsession.NewManager(llmAdapter)
// 注册子会话提供者
subManager.Register(subsession.NewGeneralProvider(personaLoader))
if memRetriever != nil {
subManager.Register(subsession.NewMemoryProvider(memRetriever))
}
if iotClient != nil {
subManager.Register(subsession.NewIoTProvider(iotClient))
}
log.Printf("子会话管理器已就绪: %d 个提供者 (%v)", len(subManager.ListProviders()), subManager.ListProviders())
// 构建新的 Orchestrator (v2.0)
orch := orchestrator.NewOrchestrator(
personaLoader,
ctxBuilder,
llmAdapter,
subManager,
memRetriever,
memExtractor,
)
log.Println("对话编排器 v2.0 已就绪")
// 注册对话API端点
mux.HandleFunc("/api/v1/chat", func(w http.ResponseWriter, r *http.Request) {
handleChat(w, r, orch, ctxBuilder, llmAdapter, personaLoader, memRetriever, memExtractor, iotClient, thinker, toolRegistry)
})
// 注册记忆API端点
mux.HandleFunc("/api/v1/memory/search", func(w http.ResponseWriter, r *http.Request) {
handleMemorySearch(w, r, memRetriever)
})
mux.HandleFunc("/api/v1/memory", func(w http.ResponseWriter, r *http.Request) {
handleMemoryCRUD(w, r, memStore, memExtractor)
})
mux.HandleFunc("/api/v1/health", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.Write([]byte(`{"status":"ok","service":"ai-core","model":"` + llmAdapter.ModelName() + `"}`))
})
// 启动HTTP服务
srv := &http.Server{
Addr: ":" + cfg.Port,
Handler: mux,
}
go func() {
log.Printf("🚀 AI-Core 服务已启动在端口 %s", cfg.Port)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatalf("服务启动失败: %v", err)
}
}()
// 优雅关闭
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Println("正在关闭 AI-Core 服务...")
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
srv.Shutdown(ctx)
log.Println("AI-Core 服务已关闭")
}
// Config AI-Core配置
type Config struct {
Port string
PersonaDir string
LLMBaseURL string
LLMAPIKey string
LLMModel string
LLMFallbackModel string
DatabaseURL string
IoTServiceURL string
AdminNickname string // 昔涟对管理员用户的基本称呼
}
func loadConfig() Config {
return Config{
Port: getEnv("AI_CORE_PORT", "8081"),
PersonaDir: getEnv("PERSONA_DIR", "./internal/persona"),
LLMBaseURL: getEnv("LLM_API_URL", "https://api.openai.com/v1"),
LLMAPIKey: getEnv("LLM_API_KEY", ""),
LLMModel: getEnv("LLM_MODEL", "gpt-4o"),
LLMFallbackModel: getEnv("LLM_FALLBACK_MODEL", "gpt-4o-mini"),
DatabaseURL: buildDatabaseURL(),
IoTServiceURL: getEnv("IOT_DEBUG_SERVICE_URL", ""),
AdminNickname: getEnv("ADMIN_NICKNAME", "管理员"),
}
}
func buildDatabaseURL() string {
host := getEnv("POSTGRES_HOST", "localhost")
port := getEnv("POSTGRES_PORT", "5432")
user := getEnv("POSTGRES_USER", "cyrene")
password := getEnv("POSTGRES_PASSWORD", "change_me")
dbname := getEnv("POSTGRES_DB", "cyrene_ai")
sslmode := getEnv("POSTGRES_SSLMODE", "disable")
return fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=%s",
user, password, host, port, dbname, sslmode)
}
func getEnv(key, fallback string) string {
if v := os.Getenv(key); v != "" {
return v
}
return fallback
}
func getEnvBool(key string, fallback bool) bool {
v := os.Getenv(key)
if v == "" {
return fallback
}
switch strings.ToLower(v) {
case "true", "1", "yes", "on":
return true
case "false", "0", "no", "off":
return false
default:
return fallback
}
}
// buildOpenAITools 将工具注册中心的定义转换为 LLM 层的 OpenAITool 格式
func buildOpenAITools(registry *tools.Registry) []llm.OpenAITool {
if registry == nil || !registry.IsEnabled() {
return nil
}
defs := registry.GetDefinitions()
if len(defs) == 0 {
return nil
}
result := make([]llm.OpenAITool, 0, len(defs))
for _, d := range defs {
result = append(result, llm.OpenAITool{
Type: "function",
Function: llm.OpenAIToolFunc{
Name: d.Name,
Description: d.Description,
Parameters: d.Parameters,
},
})
}
return result
}
// handleChat 处理对话请求(SSE 流式响应)— 使用新 Orchestrator v2.0
func handleChat(
w http.ResponseWriter,
r *http.Request,
orch *orchestrator.Orchestrator,
ctxBuilder *ctxbuild.Builder,
_ *llm.Adapter,
_ *persona.Loader,
_ *memory.Retriever,
_ *memory.Extractor,
iotClient *tools.IoTClient,
thinker *background.Thinker,
_ *tools.Registry,
) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
// 解析请求
var req struct {
UserID string `json:"user_id"`
SessionID string `json:"session_id"`
Message string `json:"message"`
Mode string `json:"mode"`
Nickname string `json:"nickname,omitempty"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "无效的请求体", http.StatusBadRequest)
return
}
if req.Mode == "" {
req.Mode = "text"
}
ctx := r.Context()
// 0. 记录用户活动(重置闲置计时器)
if thinker != nil {
thinker.RecordUserMessage()
}
// 确定用户昵称
userNickname := req.Nickname
if userNickname == "" {
userNickname = cfg.AdminNickname
}
// 0.1 缓存用户消息到会话历史
ctxBuilder.CacheMessage(req.SessionID, model.RoleUser, req.Message)
// 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("X-Accel-Buffering", "no")
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "Streaming not supported", http.StatusInternalServerError)
return
}
// 2. 调用 Orchestrator 处理(替代原有的线性处理流程)
// Orchestrator 内部处理:意图分析 → 子会话分派 → 结果汇总 → 综合生成回复
eventCh, err := orch.ProcessInput(ctx, orchestrator.ProcessParams{
UserID: req.UserID,
SessionID: req.SessionID,
Message: req.Message,
Mode: req.Mode,
Nickname: userNickname,
})
if err != nil {
errData, _ := json.Marshal(map[string]string{"delta": "", "error": fmt.Sprintf("处理失败: %v", err)})
fmt.Fprintf(w, "data: %s\n\n", errData)
flusher.Flush()
fmt.Fprintf(w, "data: [DONE]\n\n")
flusher.Flush()
return
}
messageID := fmt.Sprintf("msg-%d", time.Now().UnixNano())
// 3. 流式输出 SSE
var fullContent string
for event := range eventCh {
switch event.Type {
case model.StreamError:
log.Printf("[chat] 流式错误: %v", event.Error)
errData, _ := json.Marshal(map[string]string{"delta": "", "error": event.Error.Error()})
fmt.Fprintf(w, "data: %s\n\n", errData)
flusher.Flush()
fmt.Fprintf(w, "data: [DONE]\n\n")
flusher.Flush()
return
case model.StreamDelta:
fullContent += event.Delta
deltaData, _ := json.Marshal(map[string]string{
"delta": event.Delta,
"message_id": messageID,
})
fmt.Fprintf(w, "data: %s\n\n", deltaData)
flusher.Flush()
case model.StreamSegments:
// 发送断句信息
segData, _ := json.Marshal(map[string]interface{}{
"message_id": messageID,
"mode": req.Mode,
"segments": event.Segments,
})
fmt.Fprintf(w, "data: %s\n\n", segData)
flusher.Flush()
case model.StreamDone:
// 下发结束标记
endData, _ := json.Marshal(map[string]interface{}{
"message_id": messageID,
"mode": req.Mode,
"done": true,
})
fmt.Fprintf(w, "data: %s\n\n", endData)
flusher.Flush()
fmt.Fprintf(w, "data: [DONE]\n\n")
flusher.Flush()
}
}
// 4. 对话完成后触发昔涟的自主思考(事件驱动,非定时)
if thinker != nil {
thinker.TriggerPostChatThink()
}
}
// handleMemorySearch 处理记忆搜索请求
func handleMemorySearch(
w http.ResponseWriter,
r *http.Request,
memRetriever *memory.Retriever,
) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
userID := r.URL.Query().Get("user_id")
if userID == "" {
http.Error(w, "缺少 user_id 参数", http.StatusBadRequest)
return
}
query := r.URL.Query().Get("q")
if query == "" {
http.Error(w, "缺少 q 参数", http.StatusBadRequest)
return
}
if memRetriever == nil {
log.Printf("[memory] 记忆检索器未初始化: 数据库不可用")
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"user_id": userID,
"query": query,
"memories": []interface{}{},
"error": "记忆系统未就绪",
"errorType": "memory_store_unavailable",
"hint": "PostgreSQL 数据库不可用,请检查数据库连接配置",
})
return
}
ctx := r.Context()
memories, err := memRetriever.Retrieve(ctx, userID, query)
if err != nil {
log.Printf("[memory] 检索失败: %v", err)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"user_id": userID,
"query": query,
"memories": []interface{}{},
"error": fmt.Sprintf("检索失败: %v", err),
"errorType": "retrieve_failed",
})
return
}
if memories == nil {
memories = []memory.MemoryEntry{}
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"user_id": userID,
"query": query,
"memories": memories,
"total": len(memories),
})
}
// handleMemoryCRUD 处理记忆的 CRUD 操作
func handleMemoryCRUD(
w http.ResponseWriter,
r *http.Request,
memStore *memory.Store,
memExtractor *memory.Extractor,
) {
switch r.Method {
case http.MethodGet:
// 列出用户的所有记忆
userID := r.URL.Query().Get("user_id")
if userID == "" {
http.Error(w, "缺少 user_id 参数", http.StatusBadRequest)
return
}
if memStore == nil {
log.Printf("[memory] 记忆存储未初始化: 数据库不可用")
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"user_id": userID,
"memories": []interface{}{},
"error": "记忆系统未就绪",
"errorType": "memory_store_unavailable",
"hint": "PostgreSQL 数据库不可用,请检查数据库连接配置",
})
return
}
ctx := r.Context()
memories, err := memStore.Query(ctx, model.MemoryQuery{
UserID: userID,
Limit: 50,
})
if err != nil {
log.Printf("[memory] 查询失败: %v", err)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"user_id": userID,
"memories": []interface{}{},
"error": fmt.Sprintf("查询失败: %v", err),
"errorType": "query_failed",
})
return
}
if memories == nil {
memories = []model.MemoryEntry{}
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"user_id": userID,
"memories": memories,
"total": len(memories),
})
case http.MethodDelete:
// 删除单条记忆: DELETE /api/v1/memory?id=xxx
memoryID := r.URL.Query().Get("id")
if memoryID == "" {
http.Error(w, "缺少 id 参数", http.StatusBadRequest)
return
}
if memStore == nil {
log.Printf("[memory] 记忆存储未初始化: 无法删除")
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusServiceUnavailable)
json.NewEncoder(w).Encode(map[string]interface{}{
"error": "记忆系统未就绪",
"errorType": "memory_store_unavailable",
"hint": "PostgreSQL 数据库不可用,请检查数据库连接配置",
})
return
}
ctx := r.Context()
if err := memStore.Delete(ctx, memoryID); err != nil {
log.Printf("[memory] 删除失败: %v", err)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(map[string]interface{}{
"error": fmt.Sprintf("删除失败: %v", err),
"errorType": "delete_failed",
})
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"status": "deleted",
"memory_id": memoryID,
})
case http.MethodPost:
// 手动添加记忆
var req struct {
UserID string `json:"user_id"`
Content string `json:"content"`
Category string `json:"category"`
Priority int `json:"priority"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "无效的请求体", http.StatusBadRequest)
return
}
if req.UserID == "" || req.Content == "" {
http.Error(w, "缺少 user_id 或 content", http.StatusBadRequest)
return
}
if req.Category == "" {
req.Category = "other"
}
if req.Priority <= 0 {
req.Priority = 1
}
if memStore == nil {
log.Printf("[memory] 记忆存储未初始化: 无法保存")
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusServiceUnavailable)
json.NewEncoder(w).Encode(map[string]interface{}{
"error": "记忆系统未就绪",
"errorType": "memory_store_unavailable",
"hint": "PostgreSQL 数据库不可用,请检查数据库连接配置",
})
return
}
entry := &model.MemoryEntry{
UserID: req.UserID,
Content: req.Content,
Category: model.MemoryCategory(req.Category),
Priority: model.MemoryPriority(req.Priority),
}
ctx := r.Context()
if err := memStore.Save(ctx, entry); err != nil {
log.Printf("[memory] 保存失败: %v", err)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(map[string]interface{}{
"error": fmt.Sprintf("保存失败: %v", err),
"errorType": "save_failed",
})
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusCreated)
json.NewEncoder(w).Encode(map[string]interface{}{
"status": "saved",
"memory": entry,
})
default:
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
}
}