基于 Langflow 二开构建 AI 应用平台：技术架构与 Coze 对比分析

┌──────────────────────────────────────────────────┐
│            langflow (分发层/Distribution)          │
│        CLI、配置管理、集成打包                      │
├──────────────────────────────────────────────────┤
│          langflow-base (平台层/Platform)           │
│    API、认证、持久化、多用户、数据库                  │
├──────────────────────────────────────────────────┤
│              lfx (运行时层/Runtime)                │
│      组件系统、图引擎、执行器、基类                   │
├──────────────────────────────────────────────────┤
│         langchain-core, pydantic, SDKs            │
└──────────────────────────────────────────────────┘

# lfx.toml - 服务替换只需一行配置
[services]
database_service = "langflow.services.database.service:DatabaseService"
auth_service = "langflow.services.auth.service:AuthService"
storage_service = "langflow.services.storage.local:LocalStorageService"
cache_service = "langflow.services.cache.service:ThreadingInMemoryCache"
# 替换为你自己的实现：
# auth_service = "myapp.services.auth:EnterpriseAuthService"

# 创建自定义组件包
lfx extension init my-company-tools
cd my-company-tools
# 编写组件...
pip install -e .
# 重启 Langflow 即自动加载

                    Langflow                          Coze
            ┌──────────────────┐            ┌──────────────────┐
  优势      │  开源可控         │            │  开箱即用         │
            │  灵活编排         │            │  多渠道发布       │
            │  MCP 协议         │            │  知识库完善       │
            │  私有部署         │            │  用户体验好       │
            │  OpenAI 兼容      │            │  企业级权限       │
            ├──────────────────┤            ├──────────────────┤
  不足      │  多租户缺失       │            │  不可私有化       │
            │  权限粗糙         │            │  不可定制         │
            │  知识库薄弱       │            │  数据不可控       │
            │  发布渠道少       │            │  MCP 不支持       │
            │  用户体验偏技术   │            │  代码不可审       │
            └──────────────────┘            └──────────────────┘
                              ↓
                    二开目标：取两者之长

┌─────────────────────────────────────────────────────────────────────┐
│                        接入层 (Gateway)                              │
│   Nginx/Traefik → WAF → Rate Limit → SSL Termination               │
├─────────────────────────────────────────────────────────────────────┤
│                     前端层 (Frontend)                                │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │ 工作流   │ │ 知识库   │ │ Agent    │ │ Bot 管理 │              │
│  │ 编辑器   │ │ 管理台   │ │ 配置台   │ │ & 发布   │              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
│  React + Zustand + React Flow + MUI                                 │
├─────────────────────────────────────────────────────────────────────┤
│                     API 网关层 (API Gateway)                         │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │ V1 API   │ │ V2 API   │ │ OpenAI   │ │ MCP      │              │
│  │ (兼容)   │ │ (增强)   │ │ 兼容接口 │ │ Server   │              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
│  FastAPI + Pydantic v2                                              │
├─────────────────────────────────────────────────────────────────────┤
│                     服务层 (Services)  ← 二开核心                     │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │ 租户服务 │ │ 权限服务 │ │ 知识库   │ │ 发布服务 │              │
│  │ (新增)   │ │ (增强)   │ │ (增强)   │ │ (新增)   │              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │ 认证服务 │ │ 存储服务 │ │ 缓存服务 │ │ 追踪服务 │              │
│  │ (增强)   │ │ (保留)   │ │ (保留)   │ │ (增强)   │              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
├─────────────────────────────────────────────────────────────────────┤
│                     运行时层 (LFX Runtime)  ← 保留核心               │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │ 图引擎   │ │ 组件注册 │ │ 执行器   │ │ 沙箱     │              │
│  │ (保留)   │ │ (保留)   │ │ (保留)   │ │ (新增)   │              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
├─────────────────────────────────────────────────────────────────────┤
│                     数据层 (Data)                                    │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │ PostgreSQL│ │ Redis    │ │ S3/MinIO │ │ 向量数据库│              │
│  │ (主库)   │ │ (缓存)   │ │ (文件)   │ │ (知识库) │              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
├─────────────────────────────────────────────────────────────────────┤
│                     基础设施层 (Infrastructure)                       │
│  Docker / Kubernetes / Helm / Terraform                             │
└─────────────────────────────────────────────────────────────────────┘

优先级：配置替换 > Extension Bundle > lfx.toml 插件 > Fork 修改

my-langflow-platform/                    # 你的二开项目
├── pyproject.toml                       # 项目依赖（依赖 langflow）
├── lfx.toml                             # 服务替换配置
├── src/
│   ├── platform/                        # 平台层二开代码
│   │   ├── services/                    # 服务实现
│   │   │   ├── tenant/                  # 租户服务（新增）
│   │   │   ├── rbac/                    # RBAC 权限服务（新增）
│   │   │   ├── knowledge/               # 知识库服务（增强）
│   │   │   ├── publish/                 # 发布服务（新增）
│   │   │   ├── sandbox/                 # 代码沙箱（新增）
│   │   │   └── auth/                    # 认证增强（SSO/OAuth）
│   │   ├── api/                         # API 扩展
│   │   │   ├── v2/                      # V2 新增路由
│   │   │   └── middleware/              # 中间件（租户识别等）
│   │   └── models/                      # 数据模型扩展
│   │
│   ├── bundles/                         # 自定义组件包
│   │   ├── enterprise-tools/            # 企业工具包
│   │   ├── china-llm/                   # 国产大模型包
│   │   └── im-channels/                 # IM 渠道包
│   │
│   └── frontend/                        # 前端二开
│       ├── pages/                       # 新增页面
│       │   ├── KnowledgeBase/           # 知识库管理
│       │   ├── BotPublish/              # Bot 发布
│       │   └── Admin/                   # 管理后台
│       └── overrides/                   # 覆盖 Langflow 前端组件
│
├── deploy/                              # 部署配置
│   ├── docker-compose.yml
│   ├── helm/
│   └── terraform/
│
└── migrations/                          # 数据库迁移
    └── versions/

# platform/services/tenant/service.py
from lfx.services.base import Service
from contextvars import ContextVar

# 租户上下文变量（请求级别隔离）
_current_tenant: ContextVar["Tenant"] = ContextVar("current_tenant", default=None)

class TenantService(Service):
    """多租户服务"""
    
    name = "tenant_service"
    
    def __init__(self, db_service, settings_service):
        super().__init__()
        self.db_service = db_service
        self.settings_service = settings_service
        self.set_ready()
    
    async def get_tenant(self, tenant_id: str) -> "Tenant":
        """获取租户信息"""
        async with self.db_service.with_session() as session:
            result = await session.execute(
                select(Tenant).where(Tenant.id == tenant_id)
            )
            return result.scalar_one_or_none()
    
    async def create_tenant(self, name: str, plan: str = "free") -> "Tenant":
        """创建租户"""
        tenant = Tenant(name=name, plan=plan)
        async with self.db_service.with_session() as session:
            session.add(tenant)
            await session.commit()
            await session.refresh(tenant)
        return tenant
    
    @staticmethod
    def set_current_tenant(tenant: "Tenant"):
        """设置当前请求的租户上下文"""
        _current_tenant.set(tenant)
    
    @staticmethod
    def get_current_tenant() -> "Tenant":
        """获取当前请求的租户上下文"""
        return _current_tenant.get()

# platform/api/middleware/tenant.py
from starlette.middleware.base import BaseHTTPMiddleware

class TenantMiddleware(BaseHTTPMiddleware):
    """从请求中识别租户并注入上下文"""
    
    async def dispatch(self, request, call_next):
        # 方式1: 从子域名识别 (tenant.myapp.com)
        host = request.headers.get("host", "")
        subdomain = host.split(".")[0] if "." in host else None
        
        # 方式2: 从 Header 识别
        tenant_id = request.headers.get("X-Tenant-ID")
        
        # 方式3: 从 JWT Token 识别
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if token:
            payload = jwt_decode(token)
            tenant_id = tenant_id or payload.get("tenant_id")
        
        # 查找租户并设置上下文
        if tenant_id:
            tenant_service = get_tenant_service()
            tenant = await tenant_service.get_tenant(tenant_id)
            if tenant:
                TenantService.set_current_tenant(tenant)
        
        return await call_next(request)

# platform/models/base.py
from sqlmodel import SQLModel
from sqlalchemy import event
from sqlalchemy.orm import Session

class TenantModel(SQLModel):
    """所有需要租户隔离的模型的基类"""
    tenant_id: str = Field(index=True)
    
    @classmethod
    def query_for_current_tenant(cls, session: Session):
        """自动过滤当前租户数据"""
        tenant = TenantService.get_current_tenant()
        if tenant:
            return session.query(cls).filter(cls.tenant_id == tenant.id)
        return session.query(cls)

# 自动注入 tenant_id 的 SQLAlchemy 事件
@event.listens_for(Session, "before_flush")
def auto_inject_tenant_id(session, flush_context, instances):
    """在写入数据库前自动注入 tenant_id"""
    tenant = TenantService.get_current_tenant()
    if not tenant:
        return
    for instance in session.new:
        if isinstance(instance, TenantModel) and not instance.tenant_id:
            instance.tenant_id = tenant.id

# platform/services/rbac/service.py
import casbin
from lfx.services.authorization.base import BaseAuthorizationService

class RBACAuthorizationService(BaseAuthorizationService):
    """基于 Casbin 的 RBAC 权限服务"""
    
    SUPPORTS_CROSS_USER_FETCH = True
    
    def __init__(self, settings_service):
        super().__init__()
        self.settings_service = settings_service
        self.enforcer = casbin.AsyncEnforcer(
            "platform/services/rbac/model.conf",
            "platform/services/rbac/policy.csv",
        )
        self.set_ready()
    
    async def enforce(
        self,
        *,
        user_id: UUID,
        domain: str,    # 资源域: "flow", "knowledge", "agent", "admin"
        obj: str,       # 资源 ID
        act: str,       # 操作: "read", "write", "delete", "execute", "publish"
        context: dict | None = None,
    ) -> bool:
        """执行权限检查"""
        roles = await self._get_user_roles(user_id, domain)
        
        for role in roles:
            allowed = await self.enforcer.enforce_async(role, domain, obj, act)
            if allowed:
                return True
        
        await self._audit_log(user_id, domain, obj, act, allowed=False)
        return False

# platform/services/rbac/model.conf
[request_definition]
r = sub, dom, obj, act

[policy_definition]
p = sub, dom, obj, act

[role_definition]
g = _, _, _

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = g(r.sub, p.sub, r.dom) && r.dom == p.dom && r.obj == p.obj && r.act == p.act

[services]
authorization_service = "platform.services.rbac.service:RBACAuthorizationService"

# platform/services/knowledge/service.py
class KnowledgeService(Service):
    """知识库全生命周期管理服务"""
    
    name = "knowledge_service"
    
    async def create_knowledge_base(
        self,
        name: str,
        description: str,
        embedding_model: str,
        vector_store: str,
        chunk_strategy: dict,
        tenant_id: str,
    ) -> "KnowledgeBase":
        """创建知识库"""
        kb = KnowledgeBase(
            name=name,
            description=description,
            embedding_model=embedding_model,
            vector_store_config={
                "type": vector_store,
                "collection_name": f"kb_{uuid4().hex[:12]}",
            },
            chunk_strategy=chunk_strategy,
            tenant_id=tenant_id,
            status="active",
        )
        async with self.db_service.with_session() as session:
            session.add(kb)
            await session.commit()
        return kb
    
    async def ingest_documents(
        self, kb_id: str, files: list, metadata: dict | None = None,
    ) -> "IngestionJob":
        """
        文档入库（异步任务）
        
        流程：保存文件 → 解析文档 → 分块 → 生成嵌入 → 写入向量数据库
        """
        kb = await self.get_knowledge_base(kb_id)
        job = IngestionJob(kb_id=kb_id, status="pending", total_files=len(files))
        
        # 提交异步任务
        await self.task_service.submit(
            task_name="ingest_documents",
            task_func=self._do_ingest,
            kwargs={"kb": kb, "files": files, "metadata": metadata, "job": job},
        )
        return job
    
    async def search(
        self,
        kb_id: str,
        query: str,
        top_k: int = 5,
        search_type: str = "similarity",  # similarity / mmr / hybrid
    ) -> list[SearchResult]:
        """
        知识库检索
        
        支持三种检索策略：
        - similarity: 纯向量相似度
        - mmr: 最大边际相关性
        - hybrid: 向量 + 关键词混合检索
        """
        kb = await self.get_knowledge_base(kb_id)
        vectorstore = self._get_vectorstore(kb)
        
        if search_type == "similarity":
            results = await vectorstore.asimilarity_search_with_score(query, k=top_k)
        elif search_type == "mmr":
            results = await vectorstore.amax_marginal_relevance_search(query, k=top_k)
        elif search_type == "hybrid":
            results = await self._hybrid_search(vectorstore, query, top_k)
        
        return [SearchResult(content=doc.page_content, score=score, metadata=doc.metadata)
                for doc, score in results]

# platform/services/publish/service.py
class PublishService(Service):
    """Bot 多渠道发布服务"""
    
    CHANNEL_REGISTRY = {
        "api": APIChannel,
        "webhook": WebhookChannel,
        "wechat": WeChatChannel,      # 微信公众号/企微
        "feishu": FeishuChannel,      # 飞书
        "dingtalk": DingTalkChannel,  # 钉钉
        "slack": SlackChannel,
        "discord": DiscordChannel,
        "telegram": TelegramChannel,
        "widget": WidgetChannel,      # 网页嵌入组件
        "mcp": MCPChannel,            # MCP Server
    }
    
    async def publish(self, flow_id: str, channel: str, config: dict) -> "PublishedBot":
        """发布流程到指定渠道"""
        channel_cls = self.CHANNEL_REGISTRY[channel]
        channel_instance = channel_cls(config)
        
        flow = await self._validate_flow(flow_id)
        bot = PublishedBot(
            flow_id=flow_id,
            channel=channel,
            config=channel_instance.sanitize_config(config),
            status="active",
            endpoint_url=channel_instance.get_endpoint_url(),
        )
        await channel_instance.register(flow_id, bot)
        
        async with self.db_service.with_session() as session:
            session.add(bot)
            await session.commit()
        return bot
    
    async def handle_channel_message(self, bot_id: str, message) -> ChannelResponse:
        """
        处理来自渠道的消息
        
        流程：渠道消息格式转换 → 调用 Langflow API → 转换响应格式
        """
        bot = await self.get_bot(bot_id)
        channel_cls = self.CHANNEL_REGISTRY[bot.channel]
        channel = channel_cls(bot.config)
        
        langflow_input = channel.convert_message(message)
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"http://localhost:7860/api/v1/run/{bot.flow_id}",
                json={
                    "input_value": langflow_input,
                    "input_type": "chat",
                    "output_type": "chat",
                    "session_id": message.session_id,
                },
                headers={"x-api-key": bot.api_key},
            )
        
        return channel.convert_response(response.json())

# platform/services/publish/channels/wechat.py
class WeChatChannel:
    """微信公众号/企业微信渠道"""
    
    def __init__(self, config: dict):
        self.app_id = config["app_id"]
        self.app_secret = config["app_secret"]
        self.token = config["token"]
    
    def convert_message(self, raw) -> str:
        """微信 XML 消息 → Langflow 输入"""
        return raw.content
    
    def convert_response(self, langflow_result: dict) -> str:
        """Langflow 输出 → 微信 XML 响应"""
        reply_text = langflow_result["outputs"][0]["results"]["message"]["text"]
        return f"""<xml>
            <MsgType><![CDATA[text]]></MsgType>
            <Content><![CDATA[{reply_text}]]></Content>
        </xml>"""
    
    def get_endpoint_url(self) -> str:
        return f"/api/v1/channels/wechat/{self.app_id}"

# platform/services/sandbox/service.py
import docker
import tempfile

class SandboxService(Service):
    """代码安全执行沙箱"""
    
    SANDBOX_IMAGE = "python:3.12-slim"
    MAX_EXECUTION_TIME = 30
    MAX_MEMORY = "256m"
    
    async def execute(
        self,
        code: str,
        timeout: int = 30,
        memory_limit: str = "256m",
    ) -> SandboxResult:
        """
        在隔离容器中执行代码
        
        安全措施：无网络、内存限制、执行超时、只读文件系统、非 root
        """
        with tempfile.TemporaryDirectory() as tmpdir:
            script_path = Path(tmpdir) / "sandbox_code.py"
            script_path.write_text(code, encoding="utf-8")
            
            try:
                container = self.client.containers.run(
                    image=self.SANDBOX_IMAGE,
                    command="python /sandbox/sandbox_code.py",
                    volumes={tmpdir: {"bind": "/sandbox", "mode": "ro"}},
                    mem_limit=memory_limit,
                    network_disabled=True,
                    user="nobody",
                    read_only=True,
                    tmpfs={"/tmp": "size=64m"},
                    detach=True,
                )
                
                result = container.wait(timeout=timeout)
                stdout = container.logs(stdout=True, stderr=False)
                stderr = container.logs(stdout=False, stderr=True)
                
                return SandboxResult(
                    exit_code=result["StatusCode"],
                    stdout=stdout.decode("utf-8"),
                    stderr=stderr.decode("utf-8"),
                    timed_out=False,
                )
            except docker.errors.APIError as e:
                if "timeout" in str(e).lower():
                    return SandboxResult(exit_code=-1, stderr="Timed out", timed_out=True)
                raise
            finally:
                container.remove(force=True)

-- 启用 RLS
ALTER TABLE flow ENABLE ROW LEVEL SECURITY;
ALTER TABLE knowledge_base ENABLE ROW LEVEL SECURITY;

-- 创建租户隔离策略
CREATE POLICY tenant_isolation ON flow
    USING (tenant_id = current_setting('app.current_tenant_id')::UUID);

-- 应用层设置租户上下文
-- SET app.current_tenant_id = 'tenant-uuid-here';

# platform/services/auth/sso.py
from authlib.integrations.starlette_client import OAuth

class SSOAuthService:
    """SSO 认证服务（OAuth2/OIDC）"""
    
    def __init__(self, settings_service):
        self.oauth = OAuth()
        
        self.oauth.register(
            name="google",
            client_id=settings_service.auth_settings.GOOGLE_CLIENT_ID,
            client_secret=settings_service.auth_settings.GOOGLE_CLIENT_SECRET,
            server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
            client_kwargs={"scope": "openid email profile"},
        )
        
        self.oauth.register(
            name="github",
            client_id=settings_service.auth_settings.GITHUB_CLIENT_ID,
            client_secret=settings_service.auth_settings.GITHUB_CLIENT_SECRET,
            access_token_url="https://github.com/login/oauth/access_token",
            authorize_url="https://github.com/login/oauth/authorize",
        )
    
    async def handle_sso_callback(self, provider: str, code: str) -> User:
        """处理 SSO 回调，查找或创建用户"""
        oauth_client = self.oauth.create_client(provider)
        token = await oauth_client.authorize_access_token(code)
        userinfo = token.get("userinfo") or await oauth_client.userinfo(token=token)
        user = await self._find_or_create_user(provider, userinfo)
        return user, self._create_access_token(user)

┌─────────────────────────────────────────────────────┐
│                   知识库管理界面                       │
│  创建知识库 → 上传文档 → 配置分块策略 → 查看统计       │
├─────────────────────────────────────────────────────┤
│                   Knowledge Service                  │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ 文档解析 │ │ 分块策略 │ │ 检索策略 │            │
│  │ PDF/Word │ │ 递归/语义│ │ 向量/混合│            │
│  │ MD/HTML  │ │ 固定/自定义│ │ MMR/重排 │            │
│  └──────────┘ └──────────┘ └──────────┘            │
├─────────────────────────────────────────────────────┤
│                   向量存储层                          │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ Chroma   │ │ PGVector │ │ Qdrant   │            │
│  │ (开发)   │ │ (推荐)   │ │ (生产)   │            │
│  └──────────┘ └──────────┘ └──────────┘            │
├─────────────────────────────────────────────────────┤
│                   嵌入模型层                          │
│  OpenAI / Cohere / BGE / 本地模型                    │
└─────────────────────────────────────────────────────┘

# platform/services/knowledge/hybrid_search.py
class HybridSearchEngine:
    """混合检索引擎：向量检索 + BM25 关键词检索 + 重排"""
    
    async def search(
        self, query: str, top_k: int = 5, alpha: float = 0.7,
    ) -> list[SearchResult]:
        """
        混合检索
        1. 向量检索 top_k * 2 候选
        2. BM25 检索 top_k * 2 候选
        3. 倒数秩融合（RRF）合并
        4. 可选：重排模型精排
        """
        vector_results = await self.vectorstore.asimilarity_search_with_score(
            query, k=top_k * 2
        )
        bm25_results = self.bm25_index.search(query, top_k=top_k * 2)
        fused = self._reciprocal_rank_fusion(vector_results, bm25_results, alpha=alpha)
        
        if self.reranker:
            fused = await self.reranker.rerank(query, fused, top_k=top_k)
        
        return fused[:top_k]
    
    def _reciprocal_rank_fusion(self, vector_results, bm25_results, alpha=0.7, k=60):
        """倒数秩融合算法"""
        scores = {}
        for rank, (doc, _) in enumerate(vector_results):
            doc_id = doc.metadata.get("chunk_id", str(hash(doc.page_content)))
            scores[doc_id] = scores.get(doc_id, 0) + alpha / (k + rank + 1)
        for rank, (doc, _) in enumerate(bm25_results):
            doc_id = doc.metadata.get("chunk_id", str(hash(doc.page_content)))
            scores[doc_id] = scores.get(doc_id, 0) + (1 - alpha) / (k + rank + 1)
        return sorted(scores.items(), key=lambda x: x[1], reverse=True)

┌─────────────────────────────────────────────────┐
│                  Agent 编排层                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ ReAct    │  │ Plan &   │  │ Multi-   │      │
│  │ Agent    │  │ Execute  │  │ Agent    │      │
│  └──────────┘  └──────────┘  └──────────┘      │
├─────────────────────────────────────────────────┤
│                  记忆管理层（增强）                 │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ 短期记忆 │  │ 长期记忆 │  │ 工作记忆 │      │
│  │ (对话)   │  │ (向量DB) │  │ (Scratch)│      │
│  └──────────┘  └──────────┘  └──────────┘      │
├─────────────────────────────────────────────────┤
│                  工具层（扩展）                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ 内置工具 │  │ MCP 工具  │  │ 自定义   │      │
│  │ (Langflow)│  │ (外部)   │  │ (企业)   │      │
│  └──────────┘  └──────────┘  └──────────┘      │
├─────────────────────────────────────────────────┤
│                  人机协作层（新增）                  │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ 审批节点 │  │ 确认节点 │  │ 反馈节点 │      │
│  └──────────┘  └──────────┘  └──────────┘      │
└─────────────────────────────────────────────────┘

# bundles/enterprise-tools/components/human_in_loop/approval.py
class ApprovalNodeComponent(Component):
    """审批节点 - 暂停工作流等待人工审批"""
    
    display_name = "审批节点"
    description = "暂停工作流，等待人工审批后继续执行"
    icon = "CheckCircle"
    
    inputs = [
        Field(name="input_data", display_name="待审批数据", type="str"),
        Field(name="approvers", display_name="审批人列表", type="list[str]"),
        Field(name="timeout_minutes", display_name="超时时间(分钟)", type="int", default=60),
        Field(name="approval_type", display_name="审批类型", type="str",
              options=["any_one", "all", "majority"], default="any_one"),
    ]
    
    outputs = [
        Field(name="approved", display_name="已批准", type="str"),
        Field(name="rejected", display_name="已拒绝", type="str"),
    ]
    
    async def build(self) -> dict:
        approval = await self._create_approval_request(
            data=self.input_data, approvers=self.approvers,
            timeout=self.timeout_minutes, approval_type=self.approval_type,
        )
        result = await self._wait_for_approval(approval.id)
        if result.status == "approved":
            return {"approved": result.data}
        return {"rejected": result.reason}

lfx extension init china-llm

# bundles/china-llm/src/lfx_china_llm/components/china_llm/qwen.py
class QwenModel(LCBindingModel):
    """通义千问大模型组件"""
    
    display_name = "通义千问"
    description = "阿里云通义千问大语言模型"
    icon = "Qwen"
    
    inputs = [
        Field(name="model_name", display_name="模型", type="str",
              options=["qwen-max", "qwen-plus", "qwen-turbo", "qwen-long"],
              default="qwen-plus"),
        Field(name="api_key", display_name="API Key", type="str",
              password=True, required=True),
    ]
    
    def build(self) -> BaseChatModel:
        from langchain_community.chat_models import ChatTongyi
        return ChatTongyi(
            model=self.model_name,
            dashscope_api_key=self.api_key,
        )

# 类似实现：DeepSeek, ZhipuGLM, BaiduWenxin, Moonshot, Spark

src/frontend/src/
├── pages/
│   ├── Dashboard/              # 首页仪表盘
│   │   ├── StatsCards.tsx      # 统计卡片
│   │   ├── RecentFlows.tsx     # 最近流程
│   │   └── QuickActions.tsx    # 快捷操作
│   │
│   ├── KnowledgeBase/          # 知识库管理（新增）
│   │   ├── KnowledgeList.tsx   # 知识库列表
│   │   ├── KnowledgeDetail.tsx # 知识库详情
│   │   ├── DocumentUpload.tsx  # 文档上传
│   │   ├── ChunkPreview.tsx    # 分块预览
│   │   └── SearchTest.tsx      # 检索测试
│   │
│   ├── BotManager/             # Bot 管理（新增）
│   │   ├── BotList.tsx         # Bot 列表
│   │   ├── BotConfig.tsx       # Bot 配置
│   │   ├── ChannelSetup.tsx    # 渠道配置
│   │   └── BotAnalytics.tsx    # Bot 统计
│   │
│   └── Admin/                  # 管理后台（新增）
│       ├── UserManagement.tsx  # 用户管理
│       ├── TenantManagement.tsx# 租户管理
│       ├── RoleManagement.tsx  # 角色管理
│       └── AuditLog.tsx        # 审计日志

// pages/KnowledgeBase/KnowledgeDetail.tsx
export const KnowledgeDetail: React.FC<{ kbId: string }> = ({ kbId }) => {
  const [activeTab, setActiveTab] = useState("documents");
  const { data: kb } = useKnowledgeBase(kbId);
  const uploadMutation = useUploadDocuments();

  return (
    <Box>
      <KnowledgeHeader
        name={kb?.name}
        stats={{
          documents: kb?.document_count,
          chunks: kb?.total_chunks,
          vectorStore: kb?.vector_store_config?.type,
          embeddingModel: kb?.embedding_model,
        }}
      />

      <Tabs value={activeTab} onChange={(_, v) => setActiveTab(v)}>
        <Tab label="文档管理" value="documents" />
        <Tab label="检索测试" value="search" />
        <Tab label="分块预览" value="chunks" />
        <Tab label="设置" value="settings" />
      </Tabs>

      {activeTab === "documents" && (
        <DocumentManager
          kbId={kbId}
          onUpload={(files) => uploadMutation.mutate({ kbId, files })}
        />
      )}

      {activeTab === "search" && (
        <SearchTestPanel kbId={kbId} />
      )}
    </Box>
  );
};

┌──────────────────────────────────────────────────────────────┐
│                        负载均衡层                              │
│              Nginx / Traefik (SSL + 路由)                     │
├──────────────────────────────────────────────────────────────┤
│                        应用层                                 │
│  ┌────────────┐  ┌────────────┐  ┌────────────┐             │
│  │ Langflow   │  │ Langflow   │  │ Langflow   │  ×N         │
│  │ Worker 1   │  │ Worker 2   │  │ Worker 3   │             │
│  └────────────┘  └────────────┘  └────────────┘             │
├──────────────────────────────────────────────────────────────┤
│                        任务队列层                              │
│  ┌────────────┐  ┌────────────┐                              │
│  │ Celery     │  │ Redis      │                              │
│  │ Worker ×N  │  │ Broker     │                              │
│  └────────────┘  └────────────┘                              │
├──────────────────────────────────────────────────────────────┤
│                        数据层                                 │
│  ┌────────────┐  ┌────────────┐  ┌────────────┐             │
│  │ PostgreSQL │  │ Redis      │  │ MinIO/S3   │             │
│  │ (主库+只读 │  │ (缓存+会话)│  │ (文件存储) │             │
│  │  副本)     │  │            │  │            │             │
│  └────────────┘  └────────────┘  └────────────┘             │
│  ┌────────────┐                                              │
│  │ Qdrant     │  向量数据库                                   │
│  └────────────┘                                              │
├──────────────────────────────────────────────────────────────┤
│                        可观测层                                │
│  Prometheus + Grafana + ELK/Loki                             │
└──────────────────────────────────────────────────────────────┘

# docker-compose.prod.yml
services:
  traefik:
    image: traefik:v3.0
    command:
      - --providers.docker
      - --entrypoints.websecure.address=:443
      - --certificatesresolvers.le.acme.tlschallenge=true
    ports:
      - "443:443"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

  langflow:
    image: my-langflow-platform:latest
    environment:
      - LANGFLOW_DATABASE_URL=postgresql+asyncpg://langflow:${DB_PASSWORD}@db:5432/langflow
      - LANGFLOW_REDIS_URL=redis://redis:6379/0
      - LANGFLOW_CACHE_TYPE=redis
      - LANGFLOW_STORAGE_TYPE=s3
      - LANGFLOW_S3_ENDPOINT_URL=http://minio:9000
      - LANGFLOW_AUTHZ_ENABLED=true
      - LANGFLOW_AUTO_LOGIN=false
    depends_on:
      db: { condition: service_healthy }
      redis: { condition: service_healthy }
    labels:
      - traefik.enable=true
      - traefik.http.routers.langflow.rule=PathPrefix(`/api`)
      - traefik.http.routers.langflow.tls=true

  db:
    image: postgres:16
    environment:
      POSTGRES_DB: langflow
      POSTGRES_USER: langflow
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langflow"]

  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]

  minio:
    image: minio/minio:latest
    command: server /data --console-address ":9001"
    volumes:
      - miniodata:/data

  qdrant:
    image: qdrant/qdrant:latest
    volumes:
      - qdrantdata:/qdrant/storage

volumes:
  pgdata:
  miniodata:
  qdrantdata:

目标：让 Langflow 从"开发者工具"变成"可用的平台"

目标：补齐 Coze 的核心功能差距

目标：达到企业级安全标准，优化用户体验

月份     M1          M2          M3          M4          M5          M6
Phase 1  ████████████████
         多租户/RBAC/SSO/审计

Phase 2              ████████████████
                     知识库/Bot发布/渠道

Phase 3                          ████████████
                                 沙箱/RLS/体验

Phase 4                                      ████████████████████████
                                             生态/高级特性

┌─────────────────────────────────────────────────────┐
│              Langflow 上游仓库                        │
│              (github.com/langflow-ai/langflow)       │
└──────────────────────┬──────────────────────────────┘
                       │ git fetch upstream
                       ▼
┌─────────────────────────────────────────────────────┐
│              你的 Fork 仓库                           │
│              (github.com/your-org/langflow)          │
│                                                      │
│  main ─────────────────────────── 上游同步分支       │
│    │                                                 │
│    ├── platform/ ── 二开代码（独立目录）              │
│    ├── bundles/ ─── 自定义组件包                      │
│    └── patches/ ─── 对上游的补丁（最小化）            │
│                                                      │
│  同步策略：                                           │
│  1. 每周 fetch upstream/main                         │
│  2. rebase 你的 platform 分支                        │
│  3. patches 目录存放对上游的最小修改                   │
│  4. 优先通过 lfx.toml 插件机制替换而非 Fork           │
└─────────────────────────────────────────────────────┘

Coze：  "5 分钟上线一个 Bot"        → 面向非技术用户，快速验证
Langflow 二开："5 分钟上线一个可控的 Bot" → 面向技术团队，兼顾效率与控制

差异化关键词：
  🏠 私有部署 - 数据不出域
  🔓 开源可控 - 代码可审计
  🔌 MCP 协议 - 标准化互操作
  🛠️ 灵活编排 - 代码级定制
  🏢 企业级   - 多租户 + RBAC + SSO