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.11pyproject.toml 中requires-python 3.11FastAPI SQLModel ORM工作流引擎基于 LangGraphlanggraph1.2,2.0依赖管理使用 uvlockfile 为 uv.lock2.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 python3.11 conda activate BiShengVENV # 2. 安装后端依赖使用 uvlockfile 为 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 生态langchain1.3,2.0、langchain-core1.4,2.0并搭配langchain-openai、langchain-milvus、langchain-elasticsearch、langchain-anthropic等集成包工作流引擎langgraph1.2,2.0Agent 框架deepagents0.6.3灵思任务模式所依赖多数据库驱动除 MySQLpymysql/aiomysql、PostgreSQLasyncpg/psycopg2-binary外还内置了达梦数据库驱动dmPython/dmAsync/dmSQLAlchemy通过sys_platform ! darwin标记在 macOS 本地开发时跳过开发与测试可选依赖dev依赖组包含pytest9.0.3、pytest-asyncio1.3.0、ruff0.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-frontendDocker 编排文件位于 docker/docker-compose.yml另有 docker/docker-compose-ft.yml 与 docker/docker-compose-office.yml 分别对应微调与办公集成场景MySQL 配置见 docker/mysql/conf/my.cnfRedis 配置见 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__分支里以host0.0.0.0, port7860, workers1直接运行与上述命令等效。启动后可用/health探活该端点定义在 bisheng/main.pyapp.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 infoCelery 应用定义在 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_numint4启动的调度中心ScheduleCenter进程数必须大于 0--max_concurrencyint32单个进程内允许的最大并发任务数asyncio.Semaphore上限该 Worker 采用多进程 asyncio 的信号量限流架构主进程通过multiprocessing.Manager创建共享的max_concurrency/node_id代理随后start_schedule_center_process派生多个ScheduleCenterProcess每个进程内用asyncio.Semaphore(max_concurrency)控制并发任务队列基于 RedisLinsightQueue实现启动前还会执行check_and_terminate_incomplete_tasks清理上次遗留的未完成任务。前端开发服务器cd src/frontend/platform npm start -- --host 0.0.0.0Vite 开发服务器运行在 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/ # 仓储实现步骤创建模块目录在src/backend/bisheng/下创建模块目录包含api/和domain/子目录。定义路由在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)注册到全局路由在 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)承载平台内部 APIchat、knowledge、workflow、llm、linsight、tenant、permission 等三十余个模块router_rpc APIRouter(prefix/api/v2)承载对外开放的 RPC 类端点open_endpoints下的 chat、knowledge、workflow、llm、citation 等。新模块如需对外开放能力可参照open_endpoints的写法注册到 v2 总线。实现业务逻辑遵循调用链路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 ORMDAO 方法sync get_xxx / async aget_xxx新工作流节点开发工作流引擎基于 LangGraph目前支持 14 种执行节点类型。扩展新节点需要修改三个位置实现节点类、注册节点类型枚举、注册节点工厂映射。步骤创建节点目录在src/backend/bisheng/workflow/nodes/下创建节点子目录src/backend/bisheng/workflow/nodes/my_node/ ├── __init__.py └── my_node.py实现节点类继承BaseNodesrc/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 实现多模态输入。注册节点类型枚举在 src/backend/bisheng/workflow/common/node.py 的NodeType枚举中添加新类型class NodeType(Enum): # ... 现有类型 MY_NODE my_node注册节点工厂映射在 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 字符串。现有节点类型参考类型枚举值说明STARTstart工作流起始节点ENDend工作流终止节点INPUTinput用户输入节点OUTPUToutput结果输出节点FAKE_OUTPUTfake_output伪输出节点LLMllm大语言模型调用CODEcode代码执行节点CONDITIONcondition条件分支判断KNOWLEDGE_RETRIEVERknowledge_retriever知识库向量检索QA_RETRIEVERqa_retriever问答检索RAGrag检索增强生成TOOLtool工具调用AGENTagentAgent 智能体REPORTreport报告生成此外node.py 中还定义了NOTE note类型的注释节点仅用于画布上的展示说明不参与实际执行。每个节点类都可选的parse_log()方法返回结构化日志tool/variable/params三种日志类型供前端渲染执行过程。新 API 端点开发步骤创建端点文件在对应模块的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_modelUnifiedResponseModel) 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(datadata)认证依赖注入通过UserPayload Depends(UserPayload.get_login_user)获取当前登录用户。UserPayload定义于 src/backend/bisheng/common/dependencies/user_deps.py继承自 auth.py 的LoginUser从 JWT Cookie 中解析用户身份提供以下属性和方法属性/方法类型说明user_idint用户 IDuser_namestr用户名user_roleList[int]用户角色 ID 列表tenant_idint当前租户 ID默认 1token_versionintJWT 失效计数器版本升级后旧 token 自动失效is_global_superbool是否全局超级管理员system:global#super_adminis_admin()bool是否管理员access_check(owner_id, target_id, access_type)bool资源权限检查UserPayload还提供了租户相关的扩展依赖get_visible_tenants()返回用户可见租户集合MVP 双层规则{叶子} ∪ {根}优先读CustomMiddleware注入的visible_tenant_idsContextVarget_tenant_admin_user用于校验全局超管或当前租户子管理员否则抛 403 错误码 19801。WebSocket 端点使用UserPayload.get_login_user_from_ws变体。统一响应格式所有 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(code500, message操作失败) # 返回: {status_code: 500, status_message: 操作失败, data: null}UnifiedResponseModel是泛型BaseModelstatus_code: int、status_message: str、data: DataT。除resp_200/resp_500外同文件还提供了分页模型PageList/PageData、游标分页信封PageInfiniteCursorData用于 ReBAC 高流量列表跳过total计数前端通过next_cursor滚动加载、SSE 响应模型SSEResponseeventdata的 Server-Sent Events 格式。开发列表类接口时优先选用PageData或PageInfiniteCursorData保持全站一致。注册路由在模块的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_*.pyasyncio_mode autoasync 测试函数无需显式装饰器自定义标记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 . --fixRuff 配置同样沉淀在 pyproject.tomltarget-version py311、line-length 120启用E/W/F/I/B/C4/UP/RUF规则集isort 规则将bisheng、bisheng_langchain识别为 first-party并针对项目实际忽略E501行长交给格式化器、B008FastAPIDepends默认参数模式、RUF012SQLModel 可变类属性等规则。后端编码约定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 注册——CustomMiddlewareJWT 解码 注入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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考