上一篇 下一篇 分享链接 返回 返回顶部

企业内网搭建 Coze 智能体平台:私有化部署架构与源码实践

发布人:慈云数据-客服中心 发布时间:15小时前 阅读量:5

Coze 私有化部署方案|附源码

适用对象:希望在企业内网、政企专有云、私有云或自建服务器中部署类 Coze 智能体平台的技术团队。
目标:构建一个可私有化运行的 AI Agent / Bot 平台,支持模型接入、知识库、插件工具、工作流、会话管理、权限控制与日志审计。
说明:本文以“Coze 私有化部署方案”为主题,提供一套可落地的架构设计与参考源码。实际生产环境可根据企业安全规范、模型供应商、数据库选型和业务规模进行调整。


一、为什么企业需要 Coze 私有化部署?

Coze 是一种面向智能体应用构建的平台形态,通常具备 Bot 创建、Prompt 编排、插件调用、知识库问答、工作流自动化、多渠道发布等能力。对于个人开发者或小团队而言,使用公有云平台即可快速搭建 AI 应用;但对于企业来说,尤其是金融、政务、医疗、制造、能源等行业,私有化部署往往是刚需。

主要原因包括:

  1. 数据安全要求高
    企业内部知识库、客户数据、业务文档、合同资料、工单记录等不能随意上传至公网平台。

  2. 满足合规与审计要求
    很多行业要求系统具备访问日志、操作审计、数据留存、权限控制和敏感信息脱敏能力。

  3. 模型接入更灵活
    私有化部署可以同时接入 OpenAI、Azure OpenAI、通义千问、智谱、Claude、DeepSeek、Ollama、本地大模型等。

  4. 可与内部系统深度集成
    例如 OA、CRM、ERP、知识库、工单系统、企业微信、钉钉、飞书、内部 API 网关等。

  5. 降低长期成本
    对于高频调用场景,使用自有模型或本地推理服务可以显著降低 token 成本。

  6. 支持二次开发
    企业可以根据自身业务改造插件、工作流、权限体系、审计机制和多租户管理能力。


二、整体架构设计

一个完整的 Coze 私有化部署方案,建议拆分为以下几个核心模块:

┌──────────────────────────────────────────────┐
│                 用户访问层                    │
│   Web 控制台 / H5 / 企业微信 / 钉钉 / API      │
└───────────────────────┬──────────────────────┘
                        │
┌───────────────────────▼──────────────────────┐
│                 网关与鉴权层                  │
│       Nginx / API Gateway / OAuth2 / JWT      │
└───────────────────────┬──────────────────────┘
                        │
┌───────────────────────▼──────────────────────┐
│                Coze 应用服务层                │
│ Bot 管理 / Prompt 编排 / 工作流 / 插件 / 会话  │
└───────────────┬───────────────┬──────────────┘
                │               │
┌───────────────▼──────┐ ┌──────▼────────────────┐
│      模型服务层       │ │       知识库服务层       │
│ OpenAI / Qwen / LLM   │ │ Embedding / 向量检索     │
│ Ollama / vLLM / etc.  │ │ Milvus / pgvector / ES   │
└───────────────┬──────┘ └──────┬────────────────┘
                │               │
┌───────────────▼───────────────▼──────────────┐
│                 数据存储层                    │
│ PostgreSQL / Redis / MinIO / Elasticsearch    │
└──────────────────────────────────────────────┘

1. 前端控制台

前端控制台主要用于:

  • 创建和管理 Bot;
  • 配置 Prompt;
  • 绑定知识库;
  • 配置插件工具;
  • 设计工作流;
  • 查看会话日志;
  • 管理用户和权限;
  • 调试智能体效果。

可使用 React、Vue、Next.js 等技术栈实现。

2. 后端应用服务

后端服务是整个私有化系统的核心,负责:

  • Bot 配置管理;
  • 用户认证与权限校验;
  • 会话上下文管理;
  • 大模型调用封装;
  • 插件工具调用;
  • 知识库检索;
  • 工作流执行;
  • 日志记录与审计。

推荐技术栈:

  • Python:FastAPI / Flask;
  • Node.js:NestJS / Express;
  • Java:Spring Boot;
  • Go:Gin / Fiber。

本文源码示例主要使用 Python + FastAPI

3. 模型服务层

模型服务层可以按需选择:

模型类型 示例
公有云大模型 OpenAI、Azure OpenAI、Claude、Gemini
国内大模型 通义千问、智谱 GLM、DeepSeek、Moonshot、百川
私有化模型 Qwen、DeepSeek、Llama、ChatGLM、Yi
推理框架 Ollama、vLLM、LMDeploy、TensorRT-LLM

为了实现可插拔,建议后端统一封装一个 LLM Provider 接口,不同模型供应商实现不同 Adapter。

4. 知识库服务层

知识库通常包含以下流程:

  1. 文档上传;
  2. 文档解析;
  3. 文本切分;
  4. Embedding 向量化;
  5. 向量入库;
  6. 用户提问时向量召回;
  7. 结合上下文生成答案。

可选技术组件:

  • 向量数据库:Milvus、Weaviate、Qdrant、pgvector;
  • 文档解析:Unstructured、PyMuPDF、python-docx;
  • 对象存储:MinIO;
  • 搜索增强:Elasticsearch;
  • Embedding 模型:text-embedding-3-small、bge-m3、m3e、gte、qwen embedding。

5. 数据存储层

推荐最小化部署组合:

组件 用途
PostgreSQL 用户、Bot、配置、会话、日志
Redis 缓存、会话状态、任务队列
MinIO 文件、图片、文档存储
Qdrant / Milvus 向量检索
Nginx 反向代理与 HTTPS

三、服务器配置建议

不同规模的私有化部署可以参考以下配置。

1. 测试环境

适合 PoC、内部演示、小团队试用。

CPU:4 核
内存:8GB
磁盘:100GB SSD
GPU:可选
系统:Ubuntu 22.04 LTS

如果使用外部大模型 API,测试环境不需要 GPU。

2. 中小型生产环境

适合几十到几百名员工使用。

CPU:8-16 核
内存:32GB-64GB
磁盘:500GB SSD
GPU:可选,视是否本地部署模型而定
系统:Ubuntu 22.04 LTS / Rocky Linux 9

3. 本地大模型部署环境

如果需要部署 Qwen、DeepSeek、Llama 等本地模型,需要配置 GPU。

CPU:16 核以上
内存:64GB 以上
GPU:NVIDIA A10 / A100 / L20 / H800 等
显存:24GB-80GB,视模型参数量而定
磁盘:1TB NVMe SSD

四、部署目录结构

建议项目目录如下:

coze-private/
├── backend/
│   ├── app/
│   │   ├── main.py
│   │   ├── config.py
│   │   ├── database.py
│   │   ├── models.py
│   │   ├── routers/
│   │   │   ├── bot.py
│   │   │   ├── chat.py
│   │   │   └── knowledge.py
│   │   ├── services/
│   │   │   ├── llm.py
│   │   │   ├── rag.py
│   │   │   └── plugin.py
│   │   └── utils/
│   │       └── auth.py
│   ├── requirements.txt
│   └── Dockerfile
├── frontend/
│   └── ...
├── nginx/
│   └── nginx.conf
├── docker-compose.yml
├── .env
└── README.md

五、Docker Compose 私有化部署源码

下面提供一份可用于私有化部署的 docker-compose.yml 示例。

version: "3.9"

services:
  postgres:
    image: postgres:15
    container_name: coze-postgres
    restart: always
    environment:
      POSTGRES_USER: coze
      POSTGRES_PASSWORD: coze_password
      POSTGRES_DB: coze_private
    volumes:
      - ./data/postgres:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  redis:
    image: redis:7
    container_name: coze-redis
    restart: always
    ports:
      - "6379:6379"

  qdrant:
    image: qdrant/qdrant:latest
    container_name: coze-qdrant
    restart: always
    volumes:
      - ./data/qdrant:/qdrant/storage
    ports:
      - "6333:6333"

  minio:
    image: minio/minio:latest
    container_name: coze-minio
    restart: always
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin123
    volumes:
      - ./data/minio:/data
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"

  backend:
    build: ./backend
    container_name: coze-backend
    restart: always
    env_file:
      - .env
    depends_on:
      - postgres
      - redis
      - qdrant
      - minio
    ports:
      - "8000:8000"

  nginx:
    image: nginx:latest
    container_name: coze-nginx
    restart: always
    depends_on:
      - backend
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf
    ports:
      - "80:80"

六、环境变量配置

.env 示例:

APP_NAME=Coze Private
APP_ENV=production
SECRET_KEY=please_change_this_secret

DATABASE_URL=postgresql://coze:coze_password@postgres:5432/coze_private
REDIS_URL=redis://redis:6379/0

QDRANT_URL=http://qdrant:6333
QDRANT_COLLECTION=coze_knowledge

MINIO_ENDPOINT=minio:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin123
MINIO_BUCKET=coze-files

LLM_PROVIDER=openai
OPENAI_API_KEY=sk-xxxx
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

EMBEDDING_MODEL=text-embedding-3-small

如果企业使用本地 Ollama,可改为:

LLM_PROVIDER=ollama
OLLAMA_BASE_URL=http://host.docker.internal:11434
OLLAMA_MODEL=qwen2.5:7b

七、后端 FastAPI 源码

1. requirements.txt

fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.8.2
sqlalchemy==2.0.32
psycopg2-binary==2.9.9
python-dotenv==1.0.1
requests==2.32.3
redis==5.0.8
qdrant-client==1.11.1
openai==1.42.0
python-multipart==0.0.9

2. config.py

import os
from dotenv import load_dotenv

load_dotenv()

class Settings:
    APP_NAME = os.getenv("APP_NAME", "Coze Private")
    APP_ENV = os.getenv("APP_ENV", "development")
    SECRET_KEY = os.getenv("SECRET_KEY", "change-me")

    DATABASE_URL = os.getenv("DATABASE_URL")
    REDIS_URL = os.getenv("REDIS_URL")

    LLM_PROVIDER = os.getenv("LLM_PROVIDER", "openai")

    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

    OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
    OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5:7b")

    QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333")
    QDRANT_COLLECTION = os.getenv("QDRANT_COLLECTION", "coze_knowledge")

settings = Settings()

3. main.py

from fastapi import FastAPI
from app.routers import bot, chat, knowledge

app = FastAPI(title="Coze Private API")

app.include_router(bot.router, prefix="/api/bots", tags=["Bots"])
app.include_router(chat.router, prefix="/api/chat", tags=["Chat"])
app.include_router(knowledge.router, prefix="/api/knowledge", tags=["Knowledge"])

@app.get("/health")
def health_check():
    return {"status": "ok"}

八、Bot 管理源码

1. models.py

from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean
from sqlalchemy.sql import func
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Bot(Base):
    __tablename__ = "bots"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(128), nullable=False)
    description = Column(Text, nullable=True)
    system_prompt = Column(Text, nullable=False)
    model = Column(String(128), nullable=True)
    enabled = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

2. database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.config import settings

engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. routers/bot.py

from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Bot

router = APIRouter()

class BotCreate(BaseModel):
    name: str
    description: str | None = None
    system_prompt: str
    model: str | None = None

@router.post("/")
def create_bot(payload: BotCreate, db: Session = Depends(get_db)):
    bot = Bot(
        name=payload.name,
        description=payload.description,
        system_prompt=payload.system_prompt,
        model=payload.model
    )
    db.add(bot)
    db.commit()
    db.refresh(bot)
    return {
        "id": bot.id,
        "name": bot.name,
        "description": bot.description,
        "system_prompt": bot.system_prompt,
        "model": bot.model
    }

@router.get("/")
def list_bots(db: Session = Depends(get_db)):
    bots = db.query(Bot).filter(Bot.enabled == True).all()
    return [
        {
            "id": bot.id,
            "name": bot.name,
            "description": bot.description,
            "model": bot.model
        }
        for bot in bots
    ]

@router.get("/{bot_id}")
def get_bot(bot_id: int, db: Session = Depends(get_db)):
    bot = db.query(Bot).filter(Bot.id == bot_id).first()
    if not bot:
        return {"error": "bot not found"}
    return {
        "id": bot.id,
        "name": bot.name,
        "description": bot.description,
        "system_prompt": bot.system_prompt,
        "model": bot.model
    }

九、大模型调用封装源码

为了让平台支持多模型,建议统一封装 LLM 接口。

services/llm.py

import requests
from openai import OpenAI
from app.config import settings

class LLMService:
    def chat(self, messages: list[dict], model: str | None = None) -> str:
        if settings.LLM_PROVIDER == "openai":
            return self._chat_openai(messages, model)
        if settings.LLM_PROVIDER == "ollama":
            return self._chat_ollama(messages, model)
        raise ValueError("Unsupported LLM provider")

    def _chat_openai(self, messages: list[dict], model: str | None = None) -> str:
        client = OpenAI(
            api_key=settings.OPENAI_API_KEY,
            base_url=settings.OPENAI_BASE_URL
        )

        response = client.chat.completions.create(
            model=model or settings.OPENAI_MODEL,
            messages=messages,
            temperature=0.7
        )

        return response.choices[0].message.content

    def _chat_ollama(self, messages: list[dict], model: str | None = None) -> str:
        prompt = ""
        for msg in messages:
            role = msg.get("role", "user")
            content = msg.get("content", "")
            prompt += f"{role}: {content}\n"

        response = requests.post(
            f"{settings.OLLAMA_BASE_URL}/api/generate",
            json={
                "model": model or settings.OLLAMA_MODEL,
                "prompt": prompt,
                "stream": False
            },
            timeout=120
        )
        response.raise_for_status()
        return response.json().get("response", "")

这段代码的好处是:未来如果要接入通义千问、DeepSeek、智谱、Claude,只需要新增对应的 Provider 方法即可,不影响业务层代码。


十、聊天接口源码

routers/chat.py

from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Bot
from app.services.llm import LLMService

router = APIRouter()

class ChatRequest(BaseModel):
    bot_id: int
    message: str

@router.post("/")
def chat(payload: ChatRequest, db: Session = Depends(get_db)):
    bot = db.query(Bot).filter(Bot.id == payload.bot_id).first()

    if not bot:
        return {"error": "bot not found"}

    messages = [
        {
            "role": "system",
            "content": bot.system_prompt
        },
        {
            "role": "user",
            "content": payload.message
        }
    ]

    llm = LLMService()
    answer = llm.chat(messages, model=bot.model)

    return {
        "bot_id": bot.id,
        "question": payload.message,
        "answer": answer
    }

调用示例:

curl -X POST http://localhost:8000/api/chat/ \
  -H "Content-Type: application/json" \
  -d '{
    "bot_id": 1,
    "message": "请介绍一下公司的报销流程"
  }'

十一、知识库 RAG 源码

私有化部署中,知识库是最重要的能力之一。下面提供一个简化版 RAG 实现。

1. services/rag.py

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from openai import OpenAI
from app.config import settings
import uuid

class RAGService:
    def __init__(self):
        self.client = QdrantClient(url=settings.QDRANT_URL)
        self.embedding_client = OpenAI(
            api_key=settings.OPENAI_API_KEY,
            base_url=settings.OPENAI_BASE_URL
        )

    def init_collection(self):
        collections = self.client.get_collections().collections
        names = [c.name for c in collections]

        if settings.QDRANT_COLLECTION not in names:
            self.client.create_collection(
                collection_name=settings.QDRANT_COLLECTION,
                vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
            )

    def embed(self, text: str) -> list[float]:
        response = self.embedding_client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def add_text(self, text: str, metadata: dict | None = None):
        self.init_collection()
        vector = self.embed(text)

        point = PointStruct(
            id=str(uuid.uuid4()),
            vector=vector,
            payload={
                "text": text,
                **(metadata or {})
            }
        )

        self.client.upsert(
            collection_name=settings.QDRANT_COLLECTION,
            points=[point]
        )

        return {"status": "ok"}

    def search(self, query: str, limit: int = 5):
        self.init_collection()
        vector = self.embed(query)

        results = self.client.search(
            collection_name=settings.QDRANT_COLLECTION,
            query_vector=vector,
            limit=limit
        )

        return [
            {
                "score": item.score,
                "text": item.payload.get("text"),
                "metadata": item.payload
            }
            for item in results
        ]

2. routers/knowledge.py

from fastapi import APIRouter
from pydantic import BaseModel
from app.services.rag import RAGService

router = APIRouter()

class KnowledgeText(BaseModel):
    text: str
    source: str | None = None

@router.post("/add")
def add_knowledge(payload: KnowledgeText):
    rag = RAGService()
    return rag.add_text(
        text=payload.text,
        metadata={
            "source": payload.source or "manual"
        }
    )

@router.get("/search")
def search_knowledge(q: str):
    rag = RAGService()
    return rag.search(q)

3. 带知识库的聊天接口改造

可以将 chat.py 改造成如下逻辑:

@router.post("/rag")
def chat_with_rag(payload: ChatRequest, db: Session = Depends(get_db)):
    bot = db.query(Bot).filter(Bot.id == payload.bot_id).first()

    if not bot:
        return {"error": "bot not found"}

    rag = RAGService()
    docs = rag.search(payload.message, limit=5)

    context = "\n\n".join([
        f"资料片段{i + 1}:{doc['text']}"
        for i, doc in enumerate(docs)
    ])

    messages = [
        {
            "role": "system",
            "content": bot.system_prompt + "\n\n请优先依据给定资料回答,如果资料不足,请明确说明。"
        },
        {
            "role": "user",
            "content": f"已知资料:\n{context}\n\n用户问题:{payload.message}"
        }
    ]

    llm = LLMService()
    answer = llm.chat(messages, model=bot.model)

    return {
        "bot_id": bot.id,
        "question": payload.message,
        "references": docs,
        "answer": answer
    }

十二、插件工具调用设计

Coze 类平台的另一个核心能力是插件。插件可以让智能体调用外部系统,例如:

  • 查询天气;
  • 查询订单;
  • 创建工单;
  • 查询库存;
  • 调用企业内部 API;
  • 发送消息通知;
  • 查询数据库;
  • 执行审批流。

下面提供一个简化插件系统源码。

services/plugin.py

import requests

class PluginService:
    def call(self, name: str, params: dict):
        if name == "weather":
            return self.weather(params)
        if name == "http_api":
            return self.http_api(params)
        raise ValueError("plugin not found")

    def weather(self, params: dict):
        city = params.get("city", "北京")
        return {
            "city": city,
            "weather": "晴",
            "temperature": "26℃",
            "note": "这是示例插件,生产环境请接入真实天气 API"
        }

    def http_api(self, params: dict):
        url = params.get("url")
        method = params.get("method", "GET").upper()
        headers = params.get("headers", {})
        body = params.get("body", {})

        if not url:
            return {"error": "url is required"}

        response = requests.request(
            method=method,
            url=url,
            headers=headers,
            json=body,
            timeout=30
        )

        return {
            "status_code": response.status_code,
            "data": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text
        }

插件系统的生产级设计不建议完全让模型自由调用任意 URL,而应加入:

  • 插件白名单;
  • 参数 Schema 校验;
  • 访问权限控制;
  • 调用频率限制;
  • 敏感字段脱敏;
  • 调用日志审计;
  • 超时和重试机制;
  • 防 SSRF 安全策略。

十三、Nginx 配置源码

nginx/nginx.conf 示例:

events {}

http {
    server {
        listen 80;
        server_name _;

        client_max_body_size 100M;

        location /api/ {
            proxy_pass http://backend:8000/api/;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        }

        location /health {
            proxy_pass http://backend:8000/health;
        }
    }
}

如果有前端项目,可以增加:

location / {
    root /usr/share/nginx/html;
    index index.html;
    try_files $uri $uri/ /index.html;
}

十四、后端 Dockerfile

backend/Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

启动命令:

docker compose up -d --build

查看服务状态:

docker compose ps

查看日志:

docker compose logs -f backend

十五、数据库初始化

需要在应用启动时创建表,可以新增 init_db.py

from app.database import engine
from app.models import Base

def init_db():
    Base.metadata.create_all(bind=engine)

if __name__ == "__main__":
    init_db()

执行:

docker compose exec backend python -m app.init_db

也可以在 FastAPI 启动事件中自动初始化:

from app.database import engine
from app.models import Base

@app.on_event("startup")
def startup():
    Base.metadata.create_all(bind=engine)

生产环境建议使用 Alembic 做数据库迁移管理。


十六、权限与安全设计

私有化部署不能只关注“能跑起来”,更要关注“安全地跑起来”。

1. 用户认证

建议支持以下方式:

  • 本地账号密码;
  • LDAP / AD;
  • OAuth2;
  • 企业微信扫码登录;
  • 钉钉扫码登录;
  • 飞书登录;
  • SSO 单点登录。

2. 权限模型

推荐使用 RBAC 权限模型:

角色 权限
超级管理员 系统配置、用户管理、模型管理
空间管理员 管理当前业务空间的 Bot 和知识库
开发者 创建 Bot、调试工作流、配置插件
普通用户 使用已发布 Bot
审计员 查看日志和审计报表

3. 数据隔离

如果平台服务多个部门或多个租户,需要设计:

  • tenant_id;
  • workspace_id;
  • bot_id;
  • knowledge_base_id;
  • user_id。

所有查询都必须带上租户或空间过滤条件,避免越权访问。

4. 日志审计

建议记录以下内容:

  • 用户登录日志;
  • Bot 创建和修改记录;
  • Prompt 修改记录;
  • 插件调用记录;
  • 知识库上传记录;
  • 大模型请求日志;
  • 敏感操作日志;
  • 异常和错误日志。

5. 敏感信息保护

包括:

  • API Key 加密存储;
  • 用户输入脱敏;
  • 日志脱敏;
  • 文件访问鉴权;
  • 内部 API 白名单;
  • 禁止模型返回敏感配置;
  • 防 Prompt Injection。

十七、生产环境优化建议

1. 服务拆分

当用户量增加后,可将服务拆分为:

  • API 服务;
  • Worker 服务;
  • 知识库解析服务;
  • Embedding 服务;
  • 工作流执行服务;
  • 插件调用服务;
  • 日志服务。

2. 异步任务

文档解析和向量化不适合同步执行,建议使用:

  • Celery;
  • RQ;
  • Dramatiq;
  • Kafka;
  • RabbitMQ。

3. 缓存优化

可以使用 Redis 缓存:

  • Bot 配置;
  • 用户权限;
  • 会话上下文;
  • 热门知识库检索结果;
  • 模型调用结果。

4. 高可用部署

生产环境建议:

  • PostgreSQL 主从或云数据库;
  • Redis Sentinel / Cluster;
  • MinIO 分布式;
  • Qdrant / Milvus 集群;
  • 多副本 API 服务;
  • Nginx / SLB 负载均衡;
  • 定期备份和恢复演练。

5. 监控告警

建议接入:

  • Prometheus;
  • Grafana;
  • Loki;
  • ELK;
  • OpenTelemetry;
  • Sentry。

监控指标包括:

  • API QPS;
  • 响应时间;
  • 模型调用耗时;
  • Token 消耗;
  • 错误率;
  • 知识库命中率;
  • 插件调用成功率;
  • 数据库连接数;
  • Redis 内存使用率。

十八、部署步骤总结

完整部署流程如下:

# 1. 克隆项目
git clone https://example.com/coze-private.git
cd coze-private

# 2. 修改环境变量
cp .env.example .env
vim .env

# 3. 启动服务
docker compose up -d --build

# 4. 初始化数据库
docker compose exec backend python -m app.init_db

# 5. 检查健康状态
curl http://localhost/health

# 6. 创建 Bot
curl -X POST http://localhost/api/bots/ \
  -H "Content-Type: application/json" \
  -d '{
    "name": "企业知识助手",
    "description": "用于回答公司制度、流程和文档问题",
    "system_prompt": "你是企业内部知识助手,请用准确、简洁、专业的中文回答问题。"
  }'

# 7. 添加知识库内容
curl -X POST http://localhost/api/knowledge/add \
  -H "Content-Type: application/json" \
  -d '{
    "text": "员工报销需要在费用发生后30天内提交申请,并上传发票、付款凭证和审批单。",
    "source": "finance_policy"
  }'

# 8. 发起 RAG 问答
curl -X POST http://localhost/api/chat/rag \
  -H "Content-Type: application/json" \
  -d '{
    "bot_id": 1,
    "message": "报销需要在多久内提交?"
  }'

十九、常见问题

1. 私有化部署是否一定需要 GPU?

不一定。如果使用 OpenAI、通义千问、DeepSeek API 等外部模型服务,服务器不需要 GPU。只有当企业希望本地部署大模型推理服务时,才需要 GPU。

2. 可以完全断网运行吗?

可以,但需要满足以下条件:

  • 本地部署大模型;
  • 本地部署 Embedding 模型;
  • 本地部署向量数据库;
  • 本地部署文档解析服务;
  • 所有依赖镜像提前离线导入;
  • 前端、后端、对象存储和数据库均在内网运行。

3. 知识库问答不准确怎么办?

可以从以下方面优化:

  • 调整文档切分策略;
  • 使用更好的 Embedding 模型;
  • 增加关键词检索和混合检索;
  • 引入 rerank 模型;
  • 优化 Prompt;
  • 增加引用来源;
  • 对低置信度回答进行拒答。

4. 插件调用安全吗?

插件调用必须做好安全限制。尤其是 HTTP API 插件,必须防止 SSRF、越权访问、敏感数据泄露和恶意请求。生产环境建议所有插件都通过企业 API 网关统一管理。


二十、结语

Coze 私有化部署的核心,不只是把一个聊天机器人系统部署到内网,而是构建一个面向企业的智能体应用平台。它需要同时具备模型接入、知识库检索、插件工具、工作流编排、权限控制、安全审计和运维监控能力。

本文提供的方案和源码适合作为企业 PoC 或二次开发基础。实际生产落地时,建议重点加强以下几个方向:

  1. 多租户与权限体系;
  2. 知识库文档解析与向量检索质量;
  3. 插件调用安全;
  4. 模型调用成本控制;
  5. 日志审计与合规;
  6. 高可用和容灾备份;
  7. 与企业现有系统集成。

如果只是快速验证,可以使用 Docker Compose 一键部署;如果要进入生产环境,则建议使用 Kubernetes、Helm、CI/CD、统一监控和集中日志平台进行标准化交付。

通过以上方案,企业可以在保证数据安全与业务可控的前提下,搭建属于自己的 Coze 私有化智能体平台。

目录结构
全文