BiSheng 开发者实战指南：环境搭建、DDD 模块扩展、工作流节点与 API 开发全流程

BiSheng 开发者实战指南：环境搭建、DDD 模块扩展、工作流节点与 API 开发全流程

【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng

本指南面向 BiSheng 开源 LLM DevOps 平台（GenAI 工作流、RAG、Agent、模型管理、评估、SFT 等能力于一体的企业级应用平台）的开发者，系统讲解从零搭建本地开发环境、启动后端/Celery/Linsight/前端服务、按 DDD 约定新增业务模块、扩展 LangGraph 工作流节点、新增 API 端点，以及测试与代码风格规范。读完本文，你将具备在 BiSheng 仓库中独立开展二次开发与功能扩展的完整实操能力，所有结论均可在当前仓库源码与配置中直接验证。

技术栈与工程布局概览

在动手之前，先明确 BiSheng 前后端的技术底座，这决定了后续所有命令与代码写法：

• 后端：Python 3.11（pyproject.toml 中requires-python = ">=3.11"），FastAPI + SQLModel ORM，工作流引擎基于 LangGraph（langgraph>=1.2,<2.0），依赖管理使用 uv（lockfile 为 uv.lock，2.4.0 版本起已从 Poetry 迁移到 uv）。
• 前端：React + TypeScript + Vite，仓库内含两个应用——src/frontend/platform（平台主应用）与src/frontend/client（客户端嵌入应用）。

仓库源码顶层目录结构如下：

src/backend/ # 后端服务（bisheng + bisheng_langchain 两个 Python 包） src/frontend/ # 前端（platform 主应用 / client 客户端应用） src/test/ # 后端测试代码 docker/ # Docker Compose 编排（MySQL/Redis/MinIO/向量库等存储服务） docs/ # 架构与开发文档

环境搭建

后端环境：conda + uv

后端要求 Python 3.11 及以上（requires-python = ">=3.11"），推荐用 conda 创建虚拟环境：

# 1. 创建 Python 3.11 虚拟环境（pyproject 要求 requires-python >=3.11） conda create --name BiShengVENV python==3.11 conda activate BiShengVENV # 2. 安装后端依赖（使用 uv，lockfile 为 uv.lock） cd src/backend uv sync --frozen --python $(which python)

uv sync --frozen会严格按照 uv.lock 锁定的版本安装依赖（--frozen表示不重新解析依赖树），并在src/backend/.venv/下创建虚拟环境。后续启动服务的所有可执行文件均通过.venv/bin/调用，例如.venv/bin/uvicorn、.venv/bin/celery、.venv/bin/pytest。

关于依赖清单，pyproject.toml 中的几个关键点值得留意：

• LangChain 1.x 生态：langchain>=1.3,<2.0、langchain-core>=1.4,<2.0，并搭配langchain-openai、langchain-milvus、langchain-elasticsearch、langchain-anthropic等集成包；
• 工作流引擎：langgraph>=1.2,<2.0；
• Agent 框架：deepagents>=0.6.3（灵思任务模式所依赖）；
• 多数据库驱动：除 MySQL（pymysql/aiomysql）、PostgreSQL（asyncpg/psycopg2-binary）外，还内置了达梦数据库驱动dmPython/dmAsync/dmSQLAlchemy（通过sys_platform != 'darwin'标记在 macOS 本地开发时跳过）；
• 开发与测试可选依赖：dev依赖组包含pytest>=9.0.3、pytest-asyncio>=1.3.0、ruff>=0.9.0，测试依赖组包含fakeredis[lua]、aiosqlite等。

前端环境

两个前端应用分别安装依赖：

# Platform 前端（主应用） cd src/frontend/platform npm install # Client 前端（客户端嵌入应用） cd src/frontend/client npm install

存储服务

存储服务（MySQL、Redis、MinIO、向量库等）通过 Docker Compose 启动，然后停止与本地开发冲突的容器（避免本机直接启动后端时端口与容器内服务冲突）：

cd docker && docker compose -p bisheng up -d docker stop bisheng-backend bisheng-backend-worker bisheng-frontend

Docker 编排文件位于 docker/docker-compose.yml（另有 docker/docker-compose-ft.yml 与 docker/docker-compose-office.yml 分别对应微调与办公集成场景），MySQL 配置见 docker/mysql/conf/my.cnf，Redis 配置见 docker/redis/redis.conf。

服务启动

后端 API 服务

cd src/backend .venv/bin/uvicorn bisheng.main:app --host 0.0.0.0 --port 7860 --workers 1 --no-access-log

本地开发建议使用--workers 1以便调试；生产环境的 Docker 容器默认使用--workers 8。从源码看，bisheng/main.py 中app = create_app()创建 FastAPI 应用，并在if __name__ == "__main__"分支里以host="0.0.0.0", port=7860, workers=1直接运行，与上述命令等效。

启动后可用/health探活，该端点定义在 bisheng/main.py：

@app.get("/health") def get_health(): return {"status": "OK"}

Celery Workers

每个 Worker 需要独立的终端窗口：

# 知识库任务 Worker（文档解析、Embedding 生成、向量写入） .venv/bin/celery -A bisheng.worker.main worker -l info -c 20 -P threads -Q knowledge_celery -n knowledge@%h # 工作流任务 Worker（工作流 DAG 执行） .venv/bin/celery -A bisheng.worker.main worker -l info -c 100 -P threads -Q workflow_celery -n workflow@%h # 定时任务调度器（遥测统计、情报同步） .venv/bin/celery -A bisheng.worker.main beat -l info

Celery 应用定义在 bisheng/worker/main.py，队列名与并发数可通过 bisheng/worker/config.py 等配置进行调整。注意知识库任务与工作流任务被路由到不同队列（knowledge_celery/workflow_celery），实现计算资源隔离。

Linsight Worker（可选）

灵思 Agent 框架使用独立的 Python 进程运行，不走 Celery 队列：

.venv/bin/python bisheng/linsight/worker.py --worker_num 4 --max_concurrency 5

从 bisheng/linsight/worker.py 的命令行解析看，两个参数的含义与默认值为：

该 Worker 采用多进程 + asyncio 的信号量限流架构：主进程通过multiprocessing.Manager创建共享的max_concurrency/node_id代理，随后start_schedule_center_process派生多个ScheduleCenterProcess，每个进程内用asyncio.Semaphore(max_concurrency)控制并发，任务队列基于 Redis（LinsightQueue）实现，启动前还会执行check_and_terminate_incomplete_tasks清理上次遗留的未完成任务。

前端开发服务器

cd src/frontend/platform npm start -- --host 0.0.0.0

Vite 开发服务器运行在 3001 端口，自动将/api/和/health请求代理到后端localhost:7860；文件服务路由（/bisheng、/tmp-dir）代理到 MinIO。代理配置可参考 src/frontend/platform/vite.config.mts 与 src/frontend/platform/nginx.conf。

新模块开发约定

后端遵循领域驱动设计（DDD）模式，新增业务模块时按固定的目录结构与调用链路组织代码。仓库中几乎所有业务模块（如bisheng/knowledge、bisheng/channel、bisheng/tenant等）都遵循这一模式，新增模块可以直接照抄既有模块的结构。

目录结构

src/backend/bisheng/<module_name>/ ├── api/ # API 层 │ ├── router.py # 路由注册（创建 APIRouter） │ ├── dependencies.py # 依赖注入（可选） │ └── endpoints/ # 端点实现 │ └── <module_name>.py # CRUD 端点函数 │ └── domain/ # 领域层 ├── models/ # 领域模型（ORM 实体） ├── schemas/ # Pydantic 数据传输对象 ├── services/ # 领域服务（核心业务逻辑） └── repositories/ # 仓储层（可选） ├── interfaces/ # 仓储接口定义 └── implementations/ # 仓储实现

步骤

1. 创建模块目录：在src/backend/bisheng/下创建模块目录，包含api/和domain/子目录。

2. 定义路由：在api/router.py中创建APIRouter，设置路由前缀和标签：

from fastapi import APIRouter from bisheng.<module_name>.api.endpoints.<module_name> import router as module_router router = APIRouter(prefix='/<module_name>', tags=['<ModuleName>']) router.include_router(module_router)

3. 注册到全局路由：在 src/backend/bisheng/api/router.py 中导入并注册路由：

from bisheng.<module_name>.api.router import router as module_router router.include_router(module_router) # 注册到 v1 路由

从全局路由源码可以看到，项目实际存在两条路由总线：router = APIRouter(prefix='/api/v1')承载平台内部 API（chat、knowledge、workflow、llm、linsight、tenant、permission 等三十余个模块），router_rpc = APIRouter(prefix='/api/v2')承载对外开放的 RPC 类端点（open_endpoints下的 chat、knowledge、workflow、llm、citation 等）。新模块如需对外开放能力，可参照open_endpoints的写法注册到 v2 总线。

4. 实现业务逻辑：遵循调用链路Router -> Endpoint -> Service -> Repository -> ORM。较简单的模块可省略 Repository 层，在 Service 中直接调用 DAO。

调用链路

api/endpoints/<module_name>.py ← 接收请求，校验参数，调用 Service | v domain/services/<service>.py ← 业务逻辑编排，事务控制 | v domain/repositories/impl/<repo>.py ← 数据访问（或直接调用 database/models/ 中的 DAO） | v database/models/<model>.py ← SQLModel ORM，DAO 方法（sync get_xxx / async aget_xxx）

新工作流节点开发

工作流引擎基于 LangGraph，目前支持 14 种执行节点类型。扩展新节点需要修改三个位置：实现节点类、注册节点类型枚举、注册节点工厂映射。

步骤

1. 创建节点目录：在src/backend/bisheng/workflow/nodes/下创建节点子目录：

src/backend/bisheng/workflow/nodes/my_node/ ├── __init__.py └── my_node.py

2. 实现节点类：继承BaseNode（src/backend/bisheng/workflow/nodes/base.py），实现_run抽象方法：

from bisheng.workflow.nodes.base import BaseNode class MyNode(BaseNode): def __init__(self, **kwargs): super().__init__(**kwargs) # 从 self.node_data 中提取节点配置参数 # 将处理后的参数存入 self.node_params def _run(self, unique_id: str): """ 节点执行逻辑。 参数: unique_id: 本次执行的唯一标识 行为: - 通过 self.graph_state.get_variable() 读取上游节点变量 - 执行业务逻辑 - 通过 self.graph_state.set_variable() 写入输出变量 - 通过 self.callback_manager 发送事件（on_node_start, on_node_end 等） """ pass

从 base.py 源码可以看到BaseNode构造函数的完整签名与关键机制：

• 构造函数参数：node_data: BaseNodeData（节点配置数据，包含类型、参数分组、描述）、workflow_id: str（所属工作流 ID）、user_id: int（触发执行的运行时用户）、graph_state: GraphState（全局变量池，管理节点间数据流）、target_edges: List[EdgeBase]（出边列表）、max_steps: int（最大执行步数，默认 50）、callback: BaseCallback（回调管理器，支持流式输出）。
• 参数预处理：init_data()会遍历node_data.group_params中的NodeGroupParams/NodeParams，把各参数key -> value深拷贝进self.node_params，节点子类可在__init__中基于node_params做二次加工。
• 执行入口run()：这是框架层面的模板方法——先检查stop_flag（用户停止）与current_step >= max_steps（超步数保护，超限抛IgnoreException），随后生成exec_id并触发callback_manager.on_node_start，调用self._run(exec_id)，把返回结果通过graph_state.set_variable(self.id, key, value)写入全局变量池，最后触发on_node_end（携带log_data与input_data）。子类只需要关心_run内的业务逻辑。
• 辅助能力：get_other_node_variable()读取其他节点变量、parse_msg_with_variables()用PromptTemplateParser做{{变量}}模板替换、get_file_base64_data()将文件（含 http/https 远程文件，走file_download缓存）转 base64、contact_file_into_prompt()将图片变量拼进 HumanMessage 实现多模态输入。

3. 注册节点类型枚举：在 src/backend/bisheng/workflow/common/node.py 的NodeType枚举中添加新类型：

class NodeType(Enum): # ... 现有类型 MY_NODE = "my_node"

4. 注册节点工厂映射：在 src/backend/bisheng/workflow/nodes/node_manage.py 的NODE_CLASS_MAP中添加映射：

from bisheng.workflow.nodes.my_node.my_node import MyNode NODE_CLASS_MAP = { # ... 现有映射 NodeType.MY_NODE.value: MyNode, }

NODE_CLASS_MAP由NodeFactory消费：get_node_class()按类型字符串取类，instance_node()实例化节点（未知类型会抛出Unknown node type异常）。因此枚举值与映射 key 必须严格对应NodeType的 value 字符串。

现有节点类型参考

此外，node.py 中还定义了NOTE = "note"类型的注释节点，仅用于画布上的展示说明，不参与实际执行。每个节点类都可选的parse_log()方法返回结构化日志（tool/variable/params三种日志类型），供前端渲染执行过程。

新 API 端点开发

步骤

1. 创建端点文件：在对应模块的api/endpoints/目录下创建文件，定义路由和处理函数：

from fastapi import APIRouter, Depends from bisheng.common.dependencies.user_deps import UserPayload from bisheng.common.schemas.api import UnifiedResponseModel, resp_200 router = APIRouter(prefix='/my-resource', tags=['MyResource']) @router.get('/', response_model=UnifiedResponseModel) async def list_resources(login_user: UserPayload = Depends(UserPayload.get_login_user)): """获取资源列表。""" # login_user 包含: user_id, user_name, user_role # login_user.is_admin() 判断是否管理员 # login_user.access_check(owner_id, target_id, access_type) 检查资源权限 data = [] return resp_200(data=data)

2. 认证依赖注入：通过UserPayload = Depends(UserPayload.get_login_user)获取当前登录用户。UserPayload（定义于 src/backend/bisheng/common/dependencies/user_deps.py，继承自 auth.py 的LoginUser）从 JWT Cookie 中解析用户身份，提供以下属性和方法：

UserPayload还提供了租户相关的扩展依赖：get_visible_tenants()返回用户可见租户集合（MVP 双层规则{叶子} ∪ {根}，优先读CustomMiddleware注入的visible_tenant_idsContextVar），get_tenant_admin_user用于校验"全局超管或当前租户子管理员"（否则抛 403 + 错误码 19801）。WebSocket 端点使用UserPayload.get_login_user_from_ws变体。

3. 统一响应格式：所有 API 返回UnifiedResponseModel（定义于 src/backend/bisheng/common/schemas/api.py），通过辅助函数构造：

from bisheng.common.schemas.api import resp_200, resp_500 # 成功响应 return resp_200(data={"id": 1, "name": "test"}) # 返回: {"status_code": 200, "status_message": "SUCCESS", "data": {...}} # 错误响应 return resp_500(code=500, message="操作失败") # 返回: {"status_code": 500, "status_message": "操作失败", "data": null}

UnifiedResponseModel是泛型BaseModel（status_code: int、status_message: str、data: DataT）。除resp_200/resp_500外，同文件还提供了分页模型PageList/PageData、游标分页信封PageInfiniteCursorData（用于 ReBAC 高流量列表，跳过total计数，前端通过next_cursor滚动加载）、SSE 响应模型SSEResponse（event+data的 Server-Sent Events 格式）。开发列表类接口时优先选用PageData或PageInfiniteCursorData保持全站一致。

4. 注册路由：在模块的api/router.py中包含端点路由，然后在 src/backend/bisheng/api/router.py 全局路由中注册（见上文"注册到全局路由"）。

错误码规范

错误码体系定义在 src/backend/bisheng/common/errcode/ 和 src/backend/bisheng/api/errcode/ 中。错误码为 5 位整数，前 3 位标识模块，后 2 位标识具体错误。继承BaseErrorCode可定义模块专属错误码，支持三种输出格式：

• return_resp()-- HTTP JSON 响应
• to_sse_event()-- SSE 事件流
• websocket_close_message()-- WebSocket 关闭消息

这与 main.py 中注册的全局异常处理器相互配合：BaseErrorCode异常会被统一转换为{"status_code": code, "status_message": message, "data": ...}的 JSON 响应。

测试

运行测试

cd src/backend # 运行全部测试 .venv/bin/pytest test/ # 运行单个测试文件 .venv/bin/pytest test/test_knowledge.py # 运行单个测试用例 .venv/bin/pytest test/test_knowledge.py::test_fn # 按关键字筛选测试 .venv/bin/pytest test/ -k "keyword"

测试配置与文件位置

测试代码位于src/backend/test/目录，测试文件命名遵循test_<module>.py约定。pyproject.toml 中[tool.pytest.ini_options]已内置相关配置：

• testpaths = ["test"]、python_files = ["test_*.py"]，即默认收集test/下所有test_*.py；
• asyncio_mode = "auto"，async 测试函数无需显式装饰器；
• 自定义标记：e2e（需要运行中后端的端到端测试，可用-m "not e2e"剔除）、slow（耗时超过 5 秒的测试）；
• filterwarnings忽略 SQLAlchemy 的 DeprecationWarning，保证输出干净。

仓库测试覆盖知识库（test/knowledge/）、工作流（test/workflow/）、租户（test/tenant/）、权限（test/permission/）、灵思（test/linsight/）、渠道（test/channel/）等数十个模块，新增功能建议同步补充对应模块目录下的测试用例。

代码风格

后端

使用 Black 格式化和 Ruff 代码检查：

cd src/backend # 代码格式化 .venv/bin/black . # 代码检查与自动修复 .venv/bin/ruff check . --fix

Ruff 配置同样沉淀在 pyproject.toml：target-version = "py311"、line-length = 120，启用E/W/F/I/B/C4/UP/RUF规则集（isort 规则将bisheng、bisheng_langchain识别为 first-party），并针对项目实际忽略E501（行长交给格式化器）、B008（FastAPIDepends默认参数模式）、RUF012（SQLModel 可变类属性）等规则。

后端编码约定

• ORM 模型：定义在database/models/中，每个文件包含 Base/Read/Create/Update schema 和 DAO 类。DAO 提供同步方法（get_xxx）和异步方法（aget_xxx）两套接口，异步场景（FastAPI 端点）统一走aget_*。
• 配置读取：运行时可变配置从数据库读取（通过ConfigService.get_all_config()），静态配置从config.yaml加载（参考 src/backend/bisheng/core/config/）。
• 日志：使用 Loguru，通过from loguru import logger导入。中间件自动注入trace_id用于链路追踪，日志配置见 bisheng/core/logger.py。
• 异步任务：耗时操作投递到 Celery 队列。知识库任务路由到knowledge_celery队列，工作流任务路由到workflow_celery队列（队列清单见 bisheng/worker/main.py）。
• 中间件顺序：注意 main.py 中 Starlette 中间件为 LIFO 注册——CustomMiddleware（JWT 解码 + 注入visible_tenant_ids）需在入站路径上先于AdminScopeMiddleware执行，因此源码中先add_middleware(AdminScopeMiddleware)再add_middleware(CustomMiddleware)，开发新增中间件时需留意这一顺序约束。CORS 白名单默认包含localhost:3000/3001/5173，可通过环境变量BISHENG_CORS_ORIGINS（逗号分隔）覆盖。

前端

• TypeScript 严格模式
• 组件使用函数式组件 + Hooks
• 状态管理优先使用 Zustand store，其次 React Context
• 国际化文本通过useTranslation()获取，支持中文、英文、日文

相关文档

• 系统架构总览 -- docs/architecture/01-architecture-overview.md
• 后端模块划分 -- docs/architecture/02-backend-modules.md
• 工作流引擎设计 -- docs/architecture/03-workflow-engine.md
• 知识库/RAG 流水线 -- docs/architecture/04-knowledge-rag.md
• 灵思 Agent 框架 -- docs/architecture/05-linsight-agent.md
• 数据模型定义 -- docs/architecture/07-data-models.md
• 部署与运维 -- docs/architecture/08-deployment.md

【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng