# Linux
docker run -d --name=maxkb --restart=always -p 8080:8080 \
  -v ~/.maxkb:/opt/maxkb registry.fit2cloud.com/maxkb/maxkb

# Windows (PowerShell)
docker run -d --name=maxkb --restart=always -p 8080:8080 `
  -v C:/maxkb:/opt/maxkb registry.fit2cloud.com/maxkb/maxkb

你是一个专业的客服助手。请根据以下已知信息回答用户的问题。
已知信息：{data}
用户问题：{question}

# main.py 的核心命令行参数
parser.add_argument('action', choices=("start", "dev", "upgrade_db", "collect_static"))
parser.add_argument("services", choices=("all", "web", "task", "celery", "local_model"))

D:\data\coder\MaxKB\
├── main.py                    # 统一入口
├── pyproject.toml             # 依赖管理（使用 uv 包管理器）
├── apps/                      # Django 应用根目录
│   ├── maxkb/                 # 核心配置
│   │   ├── settings/          # 双模式配置（web/model）
│   │   │   ├── base/
│   │   │   │   ├── web.py     # Web 服务 Django 配置
│   │   │   │   └── model.py   # 本地模型服务配置
│   │   │   ├── auth/          # 认证处理器配置
│   │   │   ├── lib.py         # Celery 配置
│   │   │   └── logging.py     # 日志配置
│   │   ├── urls/
│   │   │   └── web.py         # 全局路由（admin API + chat API）
│   │   ├── conf.py            # 配置管理器（YAML + 环境变量）
│   │   └── const.py           # 常量定义
│   ├── application/           # ★ 核心：智能体/应用模块
│   ├── knowledge/             # ★ 核心：知识库模块
│   ├── models_provider/        # ★ 核心：模型供应商适配
│   ├── chat/                  # ★ 核心：对话处理模块
│   ├── tools/                 # 工具管理模块
│   ├── trigger/               # 触发器模块
│   ├── users/                 # 用户管理模块
│   ├── system_manage/         # 系统管理（权限、资源映射）
│   ├── common/                # ★ 基础设施层（30+ 子模块）
│   ├── ops/                   # 运维（Celery 配置）
│   ├── oss/                   # 对象存储模块
│   ├── folders/               # 文件夹管理
│   ├── homepage/              # 首页统计
│   ├── local_model/           # 本地模型服务
│   └── locales/               # 国际化翻译
└── ui/                        # Vue 3 前端项目

SERVER_NAME=web         →  使用 settings/base/web.py + settings/auth/web.py
SERVER_NAME=local_model →  使用 settings/base/model.py + settings/auth/model.py

# Admin API（管理后台）
urlpatterns += [path(f'{admin_path}/api/', include([
    path('application/', include('application.urls')),
    path('knowledge/', include('knowledge.urls')),
    path('models/', include('models_provider.urls')),
    # ... 其他管理模块
]))]

# Chat API（对话接口）
urlpatterns += [path(f'{chat_path}/api/', include([
    path('application/', include('chat.urls')),
]))]

                    ┌──────────────────┐
                    │     common       │  (基础设施层)
                    │  认证/缓存/工具   │
                    └────────┬─────────┘
                             │
            ┌────────────────┼────────────────┐
            │                │                │
    ┌───────┴──────┐ ┌──────┴─────┐ ┌───────┴────────┐
    │    users     │ │  models    │ │ system_manage  │
    │  (用户认证)   │ │ _provider  │ │ (权限/资源映射)  │
    └───────┬──────┘ │ (模型适配)  │ └───────┬────────┘
            │        └──────┬─────┘         │
            │               │               │
    ┌───────┴──────┐ ┌──────┴─────┐ ┌───────┴────────┐
    │  knowledge   │ │application │ │    tools       │
    │  (知识库)     │ │ (智能体)    │ │   (工具)       │
    └───────┬──────┘ └──────┬─────┘ └───────┬────────┘
            │               │               │
            │        ┌──────┴─────┐         │
            └───────>│    chat    │<────────┘
                     │  (对话)    │
                     └──────┬─────┘
                            │
                     ┌──────┴─────┐
                     │  trigger   │
                     │ (触发器)    │
                     └────────────┘

class Application(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid7)
    name = models.CharField(max_length=128)
    type = models.CharField(max_length=256)  # SIMPLE 或 WORK_FLOW
    model = models.ForeignKey(Model, on_delete=models.SET_NULL)  # 关联 LLM
    knowledge_setting = models.JSONField()     # 知识库检索配置
    model_setting = models.JSONField()         # 提示词模板、对话轮次等
    work_flow = models.JSONField()             # 工作流定义（DAG JSON）
    # ... 还有 40+ 字段：prologue, dialogue_number, problem_optimization,
    #     mcp_enable, mcp_tool_ids, tool_enable, application_enable,
    #     skill_tool_ids, long_term_enable, tts_model, stt_model 等

{
  "nodes": [
    {"id": "node_1", "type": "start-node", "properties": {...}},
    {"id": "node_2", "type": "ai-chat-node", "properties": {...}},
    {"id": "node_3", "type": "search-knowledge-node", "properties": {...}}
  ],
  "edges": [
    {"sourceNodeId": "node_1", "targetNodeId": "node_2"},
    {"sourceNodeId": "node_1", "targetNodeId": "node_3"},
    {"sourceNodeId": "node_3", "targetNodeId": "node_2"}
  ]
}

class WorkflowManage:
    def __init__(self, flow, params, ...):
        self.flow = flow           # Workflow 对象（包含 nodes 和 edges）
        self.params = params       # 运行参数（question, chat_id 等）
        self.context = {}          # 全局上下文
        self.node_chunk_manage = NodeChunkManage(self)  # 流式 chunk 管理

    def run(self):
        """入口方法"""
        start_node = self.get_start_node()      # 找到 start-node
        self.run_chain_async(start_node)         # 异步执行

    def run_chain_async(self, node):
        """提交节点到线程池异步执行"""
        future = executor.submit(self._run_node, node)
        # 200 个工作线程的 ThreadPoolExecutor

    def _run_node(self, node):
        """执行单个节点"""
        result = node.run()                      # 调用节点的 run 方法
        self.write_context(result)               # 将结果写入上下文
        next_nodes = self.get_next_node_list(node, result)  # 推导下游节点
        for next_node in next_nodes:
            self.run_chain_async(next_node)      # 递归执行下游节点

def is_interrupt(node, step_variable, global_variable):
    return node.type == 'form-node' and not node.context.get('is_submit', False)

class INode(ABC):
    @abstractmethod
    def execute(self, **kwargs) -> NodeResult:
        """节点的核心执行逻辑，返回 NodeResult"""
        pass

    def run(self) -> NodeResult:
        """模板方法：记录开始时间 → 调用 execute → 写入上下文 → 返回结果"""
        self.context['start_time'] = time.time()
        result = self.execute(**self.get_node_params())
        return result

ResetProblemStep → SearchDatasetStep → GenerateHumanMessageStep → ChatStep

pipeline_manage_builder = PipelineManage.builder()
# 如果开启了问题优化，才添加 ResetProblemStep
if chat_info.application.problem_optimization:
    pipeline_manage_builder.append_step(BaseResetProblemStep)
# 依次添加检索、Prompt 组装、对话步骤
pipeline = (pipeline_manage_builder
    .append_step(BaseSearchDatasetStep)
    .append_step(BaseGenerateHumanMessageStep)
    .append_step(BaseChatStep)
    .add_base_to_response(base_to_response)
    .build())

Knowledge (知识库)
  ├── Document (文档)
  │     ├── Paragraph (段落) ← 文档被拆分后的每个段落
  │     │     └── Embedding (向量) ← 每个段落对应的向量表示
  │     └── Problem (问题) ← FAQ 关联问题
  │           └── ProblemParagraphMapping ← 问题与段落的映射
  ├── Tag (标签) / DocumentTag ← 文档标签
  ├── Termbase (术语表) ← 优化分词的自定义词典
  └── KnowledgeFolder ← MPTT 树形文件夹

class TaskType:
    EMBEDDING = 1          # 向量化任务
    GENERATE_PROBLEM = 2   # 问题生成任务
    SYNC = 3               # 同步任务
    TOKENIZE = 4           # 分词索引任务

# Status 类使用字符串编码，每个字符位代表一种任务类型的状态
# 状态值：0=未开始, 1=进行中, 2=成功, 3=失败, 4=撤销, 5=已撤销, n=忽略
class Status:
    # 通过字符串位置索引来管理多个任务状态
    # 例如 status_str[TaskType.EMBEDDING] 可获取向量化任务的状态

class Embedding(models.Model):
    id = models.CharField(max_length=128, primary_key=True)  # 运行时通过 uuid.uuid7() 赋值
    knowledge = models.ForeignKey(Knowledge, on_delete=models.CASCADE)  # 外键关联
    document = models.ForeignKey(Document, on_delete=models.CASCADE)
    paragraph = models.ForeignKey(Paragraph, on_delete=models.CASCADE)
    embedding = VectorField(verbose_name="向量")              # pgvector 向量字段
    search_vector = SearchVectorField(verbose_name="分词", default="")  # PostgreSQL 全文检索向量
    source_type = models.CharField()                          # 来源类型
    source_id = models.CharField(max_length=128)              # 来源 ID

def search(self, query_text, query_embedding, knowledge_id_list, ...):
    # 1. 对查询文本进行向量化
    text_embedding = embedding.embed_query(query_text)
    # 2. 通过加载 .sql 文件执行原生 SQL 查询（而非 Django ORM）
    #    这样可以充分利用 pgvector 的 HNSW 索引和自定义排序逻辑
    exec_sql, exec_params = generate_sql_by_query_dict(
        {"embedding_query": query_set},
        select_string=get_file_content(
            os.path.join(PROJECT_DIR, "apps", "knowledge", "sql", "embedding_search.sql")
        ),
        with_table_name=True,
    )
    # 3. 按知识库逐个查询（利用每个知识库的 HNSW 索引）
    embedding_model = select_list(exec_sql, exec_params)

# 自定义分词：先加载术语表中的自定义词，再用 jieba 分词
terms = list(QuerySet(Termbase).filter(knowledge_id=knowledge_id)
             .values_list("content", flat=True))
search_vector = SearchVector(Value(to_ts_vector(text, user_words=terms)))

# apps/knowledge/task/embedding.py

@app.task(base=QueueOnce, once={'graceful': True})
def embedding_by_paragraph(paragraph_id, knowledge_id, ...):
    """单段落向量化"""
    embedding_model = get_embedding_model(...)
    pg_vector.save(text, embedding_model)

@app.task(base=QueueOnce, once={'graceful': True})
def embedding_by_document(document_id, knowledge_id, ...):
    """文档级向量化：遍历所有段落"""
    paragraphs = QuerySet(Paragraph).filter(document_id=document_id)
    for paragraph in paragraphs:
        embedding_by_paragraph.delay(paragraph.id, ...)

class IModelProvider(ABC):
    @abstractmethod
    def get_model_info_manage(self):
        """返回 ModelInfoManage 对象，包含该供应商支持的所有模型信息"""
        pass

    def get_model(self, model_type, model_name, model_credential, **kwargs):
        """根据模型类型和名称，实例化并返回模型对象"""
        model_info = self.get_model_info_manage().get_model_info(model_type, model_name)
        return model_info.model_class.new_instance(model_type, model_name, model_credential, **kwargs)

    def is_valid_credential(self, model_type, model_name, model_credential, ...):
        """验证凭证有效性（通常是发送一个简单请求测试连通性）"""
        model_info = self.get_model_info_manage().get_model_info(model_type, model_name)
        return model_info.model_credential.is_valid(...)

class MaxKBBaseModel(ABC):
    @staticmethod
    @abstractmethod
    def new_instance(model_type, model_name, model_credential, **model_kwargs):
        """工厂方法：根据凭证创建具体的模型实例"""
        pass

class BaseModelCredential(ABC):
    @abstractmethod
    def is_valid(self, model_type, model_name, model_credential, ...):
        """验证凭证是否有效"""
        pass

    def encryption(self, model_credential):
        """加密敏感凭证（API Key 等）"""
        pass

impl/{provider_name}/
  ├── {provider}_model_provider.py   # 供应商类（注册支持的模型列表）
  ├── credential/                    # 各模型类型的凭证校验类
  │   ├── llm.py
  │   ├── embedding.py
  │   └── ...
  ├── model/                         # 各模型类型的实际调用类
  │   ├── llm.py                     # 封装 LangChain 的 ChatModel
  │   ├── embedding.py
  │   └── ...
  └── icon/                          # 供应商标图（SVG/PNG）

class OpenAIChatModel(MaxKBBaseModel, BaseChatOpenAI):
    @staticmethod
    def new_instance(model_type, model_name, model_credential, **kwargs):
        return OpenAIChatModel(
            model=model_name,
            openai_api_key=model_credential.get('api_key'),
            openai_api_base=model_credential.get('api_base'),
            streaming=kwargs.get('streaming', False),
            # ...
        )

class ChatSerializers(serializers.Serializer):
    def chat(self, instance: dict, base_to_response: BaseToResponse = SystemToResponse()):
        """对话主入口：获取 ChatInfo 后根据应用类型分发"""
        chat_info = self.get_chat_info()  # 从缓存/数据库获取对话上下文
        application = chat_info.application

        if application.type == ApplicationTypeChoices.SIMPLE:
            return self.chat_simple(chat_info, instance, base_to_response)
        else:
            return self.chat_work_flow(chat_info, instance, base_to_response)

    def chat_simple(self, chat_info, instance, base_to_response):
        """简易模式：构建对话流水线执行"""
        # 使用 PipelineManage.builder() 构建流水线
        # 根据是否开启问题优化决定是否添加 ResetProblemStep
        pipeline = PipelineManage.builder()...build()
        return pipeline.run(...)

    def chat_work_flow(self, chat_info, instance, base_to_response):
        """工作流模式：创建 WorkflowManage 实例执行"""
        work_flow = Workflow.new_instance(application.work_flow)
        workflow = WorkflowManage(flow=work_flow, params={...})
        if stream:
            return workflow.run_stream()
        else:
            return workflow.run_block()

POST /api/application/{api_key}/chat
Content-Type: application/json

{
    "model": "maxkb",
    "messages": [{"role": "user", "content": "你好"}],
    "stream": true
}

class Trigger(models.Model):
    name = models.CharField(max_length=128)
    trigger_type = models.CharField()     # SCHEDULED 或 EVENT
    trigger_setting = models.JSONField()  # 调度配置（cron 表达式、间隔等）
    is_active = models.BooleanField()
    user = models.ForeignKey(User)        # 创建者
    desc = models.TextField()             # 描述
    meta = models.JSONField()             # 元数据

class TriggerTask(models.Model):
    trigger = models.ForeignKey(Trigger)
    source_type = models.CharField()      # APPLICATION 或 TOOL
    source_id = models.UUIDField()        # 关联的智能体或工具 ID
    parameter = models.JSONField()        # 输入参数
    is_active = models.BooleanField()

class TaskRecord(models.Model):
    """触发器任务执行记录"""
    trigger = models.ForeignKey(Trigger)
    trigger_task = models.ForeignKey(TriggerTask)
    state = models.CharField()            # PENDING/STARTED/SUCCESS/FAILURE 等
    run_time = models.FloatField()        # 执行耗时

class ScheduledTrigger:
    def deploy(self, trigger):
        """将触发器部署到调度器"""
        config = trigger.trigger_setting
        if config['type'] == 'cron':
            scheduler.add_job(
                func=ScheduledTrigger.execute,
                trigger='cron',
                kwargs={"trigger": trigger, "trigger_task": task},
                **config['cron_params']
            )

def execute(trigger, trigger_task):
    rlock = RedisLock()
    lock_key = f'{trigger.id}:{trigger_task.source_id}'
    if rlock.try_lock(lock_key, 30 * 30):  # 锁超时 30 分钟
        try:
            # 根据 source_type 决定执行智能体对话还是工具调用
            if trigger_task.source_type == 'APPLICATION':
                # 调用智能体
                ...
            elif trigger_task.source_type == 'TOOL':
                # 调用工具
                ...
        finally:
            rlock.un_lock(lock_key)

# 1. 安装依赖（使用 uv 包管理器）
pip install uv
cd D:\data\coder\MaxKB
uv sync

# 2. 启动 PostgreSQL（需要 pgvector 扩展）
docker run -d --name=maxkb-pg -p 5432:5432 \
  -e POSTGRES_DB=maxkb -e POSTGRES_USER=maxkb -e POSTGRES_PASSWORD=maxkb \
  pgvector/pgvector:pg16

# 3. 启动 Redis
docker run -d --name=maxkb-redis -p 6379:6379 redis:7

# 4. 配置数据库连接
# 编辑 apps/maxkb/config.yaml 或设置环境变量

# 5. 启动 Web 服务（开发模式）
python main.py dev web

# 6. 启动 Celery Worker（另开终端）
python main.py dev celery

cd D:\data\coder\MaxKB\ui
npm install
npm run dev        # 启动开发服务器

http://localhost:8080/admin/api/schema/swagger-ui/

1. 用户请求到达
   POST /chat/api/application/{api_key}/chat
   Body: {"message": "MaxKB 支持哪些模型？", "stream": true}

2. Chat 模块接收请求
   ChatSerializers.chat()
   → 判断应用类型为 SIMPLE
   → 调用 chat_simple()

3. 对话流水线启动
   PipelineManage.run()

4. 问题优化（可选）
   ResetProblemStep.execute()
   → 使用 LLM 重写问题（如果开启了此功能）
   → 输出优化后的问题："MaxKB 智能体平台支持的模型供应商和模型类型有哪些？"

5. 知识库检索
   SearchDatasetStep.execute()
   → 调用 PGVector.search()
   → 文本向量化：embedding.embed_query("MaxKB 智能体平台支持的模型...")
   → 在 pgvector 中执行余弦相似度搜索
   → 同时执行全文检索（如果配置了混合检索）
   → 融合排序，返回 Top K 个段落

6. Prompt 组装
   GenerateHumanMessageStep.execute()
   → 加载提示词模板
   → 填充变量：{data} = 检索到的段落, {question} = 用户问题
   → 渲染 Jinja2 模板

7. LLM 调用
   ChatStep.execute()
   → 获取 LLM 模型实例（通过 models_provider 模块）
   → 调用 model.stream(messages) 获取流式响应
   → 逐 token 通过 SSE 推送给前端

8. 后处理
   WorkFlowPostHandler.handler()
   → 保存 ChatRecord（问题、回答、Token 消耗、运行详情）
   → 更新 Redis 对话缓存
   → 异步触发长期记忆提取（Celery 任务）

9. SSE 响应完成
   → 前端收到 [DONE] 信号
   → 渲染完整的回答