LingFlow — 系统架构

整体架构图

LingFlow 采用分层架构设计，从连接管理到服务编排，每一层都独立可扩展。

架构层次说明

连接层 (connections/wss.go)

负责 WebSocket 连接的生命周期管理。

WebSocket 连接管理（连接池、状态追踪）
心跳检测引擎（Ping/Pong 双向超时检测）
认证与鉴权（HMAC-SHA256 Token 验证）
Origin 白名单 / IP 连接数限制
帧大小限制（64KB 上限）

服务层 (services/)

核心业务逻辑编排层。

消息路由分发（ChatHandler）
流式响应编排（system_thinking + system_response）
技能检索执行（SkillExecutor + SkillRegistry）
AI 技能创建器（SkillCreator）
提示注入防护（双重正则检测）

基础设施层

AWS Bedrock LLM（Converse API 统一接口）
S3 技能存储（动态加载、热更新）
事件溯源存储（InMemoryEventStore）
聚合与投影（ChatSessionAggregate）
命令总线（CommandBus）

核心组件

WebSocketConnectionManager

管理所有 WebSocket 连接的生命周期。

go

type WebSocketConnectionManager struct {
    connections         map[string]*connectionState
    eventStore          events.EventStore
    heartbeatInterval   time.Duration
    heartbeatTimeout    time.Duration
    writeTimeout        time.Duration
    maxConnectionsPerIP int
    ipConnectionCount   map[string]int
}

连接管理器的关键常量定义如下，覆盖心跳间隔、超时阈值与并发连接限制：

go

const (
    defaultHeartbeatInterval     = 30 * time.Second
    defaultHeartbeatTimeout      = 90 * time.Second
    defaultHeartbeatWriteTimeout = 10 * time.Second
    defaultMaxConnectionsPerIP   = 10
    defaultMaxFrameBytes         = 64 * 1024 // 64KB
)

ChatHandler

通过技能增强的 LLM 管道处理传入的 WebSocket 消息。

go

type ChatHandler struct {
    executor      *SkillExecutor
    registry      *SkillRegistry
    s3Loader      *S3SkillLoader
    messageSender MessageSender
}

LLMService 接口

定义与 LLM 后端交互的统一接口，支持 Bedrock 和 Mock 两种实现。

go

type LLMService interface {
    Generate(ctx context.Context, request LLMRequest) (LLMResponse, error)
    HealthCheck(ctx context.Context) error
}

BedrockConfig

go

type BedrockConfig struct {
    Region      string
    ModelID     string
    MaxTokens   int
    Temperature float32
    TopP        float32
    Timeout     time.Duration
}

事件溯源架构

所有状态变更以不可变事件记录，支持审计追踪、会话回放、时间旅行调试。

事件类型

go

const (
    EventTypeChatSessionConnected        EventType = "chat_session_connected"
    EventTypeChatSessionDisconnected     EventType = "chat_session_disconnected"
    EventTypeChatMessageReceived         EventType = "chat_message_received"
    EventTypeChatMessageProcessed        EventType = "chat_message_processed"
    EventTypeChatMessageProcessingFailed EventType = "chat_message_processing_failed"
    EventTypeChatMessageBroadcasted      EventType = "chat_message_broadcasted"
    EventTypeHeartbeatPingReceived       EventType = "heartbeat_ping_received"
    EventTypeHeartbeatPongSent           EventType = "heartbeat_pong_sent"
    EventTypeHeartbeatTimeout            EventType = "heartbeat_timeout"
    EventTypeSkillExecutionStarted       EventType = "skill_execution_started"
    EventTypeSkillExecutionCompleted     EventType = "skill_execution_completed"
    EventTypeLLMGenerationStarted        EventType = "llm_generation_started"
    EventTypeLLMGenerationCompleted      EventType = "llm_generation_completed"
)

DomainEvent 结构

go

type DomainEvent struct {
    EventID     string            `json:"event_id"`
    StreamID    string            `json:"stream_id"`
    AggregateID string            `json:"aggregate_id"`
    EventType   EventType         `json:"event_type"`
    Version     int64             `json:"version"`
    OccurredAt  time.Time         `json:"occurred_at"`
    Data        json.RawMessage   `json:"data"`
    Metadata    map[string]string `json:"metadata,omitempty"`
}

事件溯源核心概念

概念	说明
EventStore	事件存储接口，当前使用内存实现，可替换为数据库
Aggregate Root	聚合根，封装业务不变量（ChatSessionAggregate）
Command	命令对象，触发状态变更
EventHandler	事件处理器，响应已发生的事件
Projection	投影，将事件流转换为查询视图
CommandBus	命令总线，路由命令到对应处理器

事件溯源数据流

客户端发送消息 - 生成 Command
CommandBus 路由到对应 Handler
Handler 加载 Aggregate 当前状态
Aggregate 验证业务规则并产生 Event
EventStore 持久化 Event
EventBus 广播 Event 给所有 Projection
Projection 更新查询视图
ChatHandler 通过流式 WebSocket 推送响应

消息处理流程

认证流程

项目源码结构

text

internal/
├── connections/
│   └── wss.go                    # WebSocket 连接管理器
├── events/
│   ├── event.go                  # 领域事件定义
│   ├── store.go                  # 事件存储接口
│   ├── aggregate.go              # 聚合根基类
│   ├── bus.go                    # 事件总线
│   ├── command.go                # 命令定义
│   ├── handler.go                # 命令处理器
│   ├── chat_events.go            # 聊天领域事件
│   └── chat_projection.go        # 聊天投影
├── models/
│   ├── message.go                # WebSocket 消息模型
│   ├── responses.go              # 响应模型
│   └── skills.go                 # 技能模型
├── services/
│   ├── server.go                 # 服务器启动入口
│   ├── chat_handler.go           # 聊天消息处理器
│   ├── llm.go                    # LLM 服务（Bedrock + Mock）
│   ├── s3_skill_loader.go        # S3 技能加载器
│   ├── skill_registry.go         # 技能注册中心
│   ├── skill_executor.go         # 技能执行器
│   ├── skill_creator.go          # AI 技能创建器
│   ├── auth_handler.go           # 认证处理器
│   ├── notifier.go               # 通知器
│   └── aws/lambda.go             # AWS Lambda 适配器
└── utilities/
    ├── aws_util.go               # AWS 工具函数
    ├── config_loader.go          # 配置加载器
    └── logger.go                 # 日志系统