13 KiB

Raw Permalink Blame History

Luxx 项目架构

技术栈

框架: FastAPI 0.109+
数据库: SQLAlchemy 2.0+ (同步模式)
认证: JWT (PyJWT)
HTTP客户端: httpx, requests
配置: YAML (PyYAML)
代码执行: Python 原生执行
网页爬虫:
- httpx - HTTP 客户端
- beautifulsoup4 - HTML 解析
- lxml - XML/HTML 解析器

目录结构

luxx/
├── __init__.py           # FastAPI 应用工厂
├── config.py             # 配置管理（YAML）
├── database.py           # 数据库连接
├── models.py             # ORM 模型
├── routes/               # API 路由层
│   ├── __init__.py      # 路由聚合
│   ├── auth.py          # 认证 (登录/注册)
│   ├── conversations.py # 会话管理 (CRUD)
│   ├── messages.py      # 消息处理 (流式/同步)
│   ├── providers.py     # LLM 提供商管理
│   └── tools.py         # 工具管理
├── services/            # 服务层
│   ├── chat.py         # 聊天服务 (Agentic Loop)
│   └── llm_client.py   # LLM 客户端
├── tools/              # 工具系统
│   ├── core.py         # 核心类 (ToolRegistry, ToolDefinition, ToolResult)
│   ├── factory.py      # @tool 装饰器
│   ├── executor.py     # 工具执行器 (缓存/并行)
│   ├── services.py     # 工具服务层
│   └── builtin/        # 内置工具
│       ├── __init__.py # 工具注册入口
│       ├── code.py     # 代码执行 (python_execute, python_eval)
│       ├── crawler.py  # 网页爬虫 (web_search, web_fetch, batch_fetch)
│       └── data.py     # 数据处理 (process_data)
└── utils/              # 工具函数
    └── helpers.py     # 密码哈希、ID生成、响应封装

run.py                  # 应用入口文件
config.yaml             # 配置文件

核心组件

1. 应用工厂 (`init.py`)

FastAPI 应用入口，使用 lifespan 管理生命周期：

启动：初始化数据库、注册内置工具、创建默认管理员用户
关闭：清理资源

# 默认管理员账号
username: admin
password: admin123

2. 配置管理 (`config.py`)

使用 YAML 文件管理配置：

配置文件：config.yaml
环境变量替换：${VAR_NAME}
单例模式全局访问
默认值支持

# config.yaml 示例
app:
  secret_key: ${APP_SECRET_KEY}
  debug: true

database:
  type: sqlite
  url: sqlite:///./chat.db

llm:
  provider: deepseek
  api_key: ${DEEPSEEK_API_KEY}
  api_url: https://api.deepseek.com/v1

3. 数据库 (`database.py`)

SQLAlchemy 同步支持
SQLite 默认数据库
依赖注入获取会话

4. ORM 模型 (`models.py`)

erDiagram
    USER {
        int id PK
        string username UK
        string email UK
        string password_hash
        string role
        boolean is_active
        datetime created_at
    }
    
    PROJECT {
        string id PK
        int user_id FK
        string name
        text description
        datetime created_at
        datetime updated_at
    }
    
    CONVERSATION {
        string id PK
        int user_id FK
        int provider_id FK "optional"
        string project_id FK "optional"
        string title
        string model
        text system_prompt
        float temperature
        int max_tokens
        boolean thinking_enabled
        datetime created_at
        datetime updated_at
    }
    
    MESSAGE {
        string id PK
        string conversation_id FK
        string role
        longtext content "JSON 格式"
        int token_count
        text usage "JSON 格式"
        datetime created_at
    }
    
    LLM_PROVIDER {
        int id PK
        int user_id FK
        string name
        string provider_type
        string base_url
        string api_key
        string default_model
        int max_tokens
        boolean is_default
        boolean enabled
        datetime created_at
        datetime updated_at
    }
    
    USER ||--o{ PROJECT : "has"
    USER ||--o{ CONVERSATION : "has"
    USER ||--o{ LLM_PROVIDER : "configures"
    PROJECT ||--o{ CONVERSATION : "contains"
    LLM_PROVIDER ||--o{ CONVERSATION : "uses"
    CONVERSATION ||--o{ MESSAGE : "has"

Message Content JSON 结构

content 字段统一使用 JSON 格式存储：

User 消息：

{
  "text": "用户输入的文本内容",
  "attachments": [
    {"name": "utils.py", "extension": "py", "content": "..."}
  ]
}

Assistant 消息：

{
  "text": "AI 回复的文本内容",
  "tool_calls": [...],
  "steps": [
    {"id": "step-0", "index": 0, "type": "thinking", "content": "..."},
    {"id": "step-1", "index": 1, "type": "text", "content": "..."},
    {"id": "step-2", "index": 2, "type": "tool_call", "id_ref": "call_xxx", "name": "...", "arguments": "..."},
    {"id": "step-3", "index": 3, "type": "tool_result", "id_ref": "call_xxx", "name": "...", "content": "...", "success": true}
  ]
}

steps 字段是渲染顺序的唯一数据源，按 index 顺序排列。thinking、text、tool_call、tool_result 可以在多轮迭代中穿插出现。

5. 工具系统

classDiagram
    class ToolDefinition {
        +str name
        +str description
        +dict parameters
        +Callable handler
        +str category
        +to_openai_format() dict
    }

    class ToolResult {
        +bool success
        +Any data
        +str error
        +to_dict() dict
        +ok(data) ToolResult$
        +fail(error) ToolResult$
    }

    class ToolRegistry {
        +_tools: Dict
        +register(tool) void
        +get(name) ToolDefinition?
        +list_all() List~dict~
        +list_by_category(category) List~dict~
        +execute(name, arguments) dict
        +remove(name) bool
    }

    class ToolExecutor {
        +enable_cache: bool
        +cache_ttl: int
        +max_workers: int
        +_cache: Dict
        +_call_history: List
        +process_tool_calls(tool_calls, context) list
        +process_tool_calls_parallel(tool_calls, context) list
        +clear_cache() void
        +get_history(limit) List
    }

内置工具

工具	功能	说明
`python_execute`	执行 Python 代码	支持 print 输出、变量访问
`python_eval`	计算表达式	快速求值
`web_search`	DuckDuckGo HTML	DuckDuckGo HTML 搜索
`web_fetch`	网页抓取	httpx + BeautifulSoup，支持 text/links/structured
`batch_fetch`	批量抓取	并发获取多个页面
`process_data`	数据处理	JSON 转换、格式化等

工具开发规范

所有工具必须遵循统一的开发规范，确保错误处理和返回格式一致。

核心原则：装饰器自动处理一切，工具函数只写业务逻辑

from luxx.tools.factory import tool

@tool(
    name="my_tool",
    description="工具描述",
    parameters={...},
    required_params=["arg1"],  # 自动验证
    category="my_category"
)
def my_tool(arguments: dict):
    # 业务逻辑 - 只管返回数据
    data = fetch_data(arguments["arg1"])
    return {"items": data, "count": len(data)}
    
    # 或者直接抛出异常（装饰器自动捕获并转换）
    if invalid:
        raise ValueError("Invalid input")

装饰器自动处理：

必需参数验证（required_params）
所有异常捕获和转换
结果格式统一包装

返回格式转换

工具函数返回/抛出	装饰器转换为
`return {"result": "ok"}`	`{"success": true, "data": {...}, "error": null}`
`raise ValueError("msg")`	`{"success": false, "data": null, "error": "ValueError: msg"}`
`raise Exception()`	`{"success": false, "data": null, "error": "..."}`

工具调用流程

LLM 请求
    ↓
ToolRegistry.execute(name, args)
    ↓
@tool 装饰器
    ├─ 验证 required_params
    ├─ 执行工具函数 (try-except 包裹)
    ├─ 捕获异常 → 转换为 error
    ├─ 包装返回格式
    └─ 返回 ToolResult
    ↓
ToolExecutor 返回结果
    ↓
前端 ProcessBlock 显示

6. 服务层

ChatService (`services/chat.py`)

核心聊天服务：

Agentic Loop 迭代执行（最多 10 轮）
流式 SSE 响应
工具调用编排（并行执行）
消息历史管理
自动重试机制
支持 thinking_content 提取
Token 用量追踪

LLMClient (`services/llm_client.py`)

LLM API 客户端：

多提供商：DeepSeek、GLM、OpenAI
流式/同步调用
错误处理和重试
Token 计数

7. 认证系统 (`routes/auth.py`)

JWT Bearer Token
Bcrypt 密码哈希
用户注册/登录

8. API 路由

路由	方法	说明
`/auth/register`	POST	用户注册
`/auth/login`	POST	用户登录
`/conversations`	GET	会话列表（分页）
`/conversations`	POST	创建会话
`/conversations/{id}`	GET	会话详情
`/conversations/{id}`	PUT	更新会话
`/conversations/{id}`	DELETE	删除会话
`/messages`	GET	消息列表
`/messages`	POST	发送消息（同步）
`/messages/stream`	POST	发送消息（流式 SSE）
`/messages/{id}`	DELETE	删除消息
`/providers`	GET	LLM 提供商列表
`/providers`	POST	创建提供商
`/providers/{id}`	GET	提供商详情
`/providers/{id}`	PUT	更新提供商
`/providers/{id}`	DELETE	删除提供商
`/providers/{id}/test`	POST	测试提供商连接
`/tools`	GET	可用工具列表
`/health`	GET	健康检查
`/`	GET	服务信息

数据流

消息处理流程

sequenceDiagram
    participant Client
    participant API as POST /messages/stream
    participant CS as ChatService
    participant LLM as LLM API
    participant TE as ToolExecutor
    
    Client->>API: POST {content, tools, thinking_enabled}
    API->>CS: stream_response()
    
    loop MAX_ITERATIONS (10)
        CS->>LLM: call(messages, tools)
        LLM-->>CS: SSE Stream
        
        alt tool_calls
            CS->>TE: process_tool_calls_parallel()
            TE-->>CS: tool_results
            CS->>CS: 追加到 messages
        end
    end
    
    CS->>CS: _save_message()
    CS->>API: SSE Stream
    API-->>Client: 流式响应

SSE 事件

事件	说明
`process_step`	结构化步骤（thinking/text/tool_call/tool_result），携带 `id`、`index` 确保渲染顺序
`done`	响应完成，携带 message_id、token_count、usage
`error`	错误信息

process_step 事件格式

{"type": "process_step", "step": {"id": "step-0", "index": 0, "type": "thinking", "content": "..."}}
{"type": "process_step", "step": {"id": "step-1", "index": 1, "type": "text", "content": "回复文本..."}}
{"type": "process_step", "step": {"id": "step-2", "index": 2, "type": "tool_call", "id_ref": "call_abc", "name": "web_search", "arguments": "{\"query\": \"...\"}"}}
{"type": "process_step", "step": {"id": "step-3", "index": 3, "type": "tool_result", "id_ref": "call_abc", "name": "web_search", "content": "{...}", "success": true}}

字段	说明
`id`	步骤唯一标识（格式 `step-{index}`）
`index`	步骤序号，确保按正确顺序显示
`type`	步骤类型：`thinking` / `text` / `tool_call` / `tool_result`
`id_ref`	工具调用引用 ID（仅 tool_call/tool_result）
`name`	工具名称（仅 tool_call/tool_result）
`arguments`	工具调用参数 JSON 字符串（仅 tool_call）
`content`	内容（thinking 的思考内容、text 的文本、tool_result 的返回结果）
`success`	工具执行是否成功（仅 tool_result）

done 事件格式

{"type": "done", "message_id": "uuid", "token_count": 1234, "usage": {"prompt_tokens": 100, "completion_tokens": 200, "total_tokens": 300}}

配置示例

config.yaml

app:
  secret_key: ${APP_SECRET_KEY}
  debug: true
  host: 0.0.0.0
  port: 8000

database:
  type: sqlite
  url: sqlite:///./chat.db

llm:
  provider: deepseek
  api_key: ${DEEPSEEK_API_KEY}
  api_url: https://api.deepseek.com/v1

tools:
  enable_cache: true
  cache_ttl: 300
  max_workers: 4
  max_iterations: 10

环境变量

变量	说明	示例
`APP_SECRET_KEY`	应用密钥	`your-secret-key`
`DEEPSEEK_API_KEY`	DeepSeek API	`sk-xxxx`
`DATABASE_URL`	数据库连接	`sqlite:///./chat.db`

项目结构说明

入口文件

run.py - 启动 Uvicorn 服务器

响应格式

所有 API 统一使用响应封装：

// 成功
{"success": true, "data": {...}, "message": "操作成功"}

// 错误
{"success": false, "error": "错误信息", "code": 404}

工具缓存机制

ToolExecutor 支持结果缓存：

TTL: 5 分钟（可配置）
缓存 Key: {tool_name}:{sorted_arguments_json}
调用历史记录最近 1000 条

流式响应特点

实时返回 thinking_content（模型思考过程）
实时返回 text 增量更新
工具调用串行执行，结果批量返回
最终 done 事件包含完整 message_id 和 token 用量

13 KiB Raw Permalink Blame History Unescape Escape