Enterprise_RAG_Agent_Platform：企业级 RAG 落地方案

项目背景

这是 BigGraph 的"企业版"——如果说 BigGraph 是 RAG 技术的实验场，那么 Enterprise_RAG_Agent_Platform 就是面向企业落地的最佳实践。企业级 RAG 系统面临的挑战不仅仅是"检索准不准"，还包括：多租户隔离、细粒度权限控制、审计日志、服务高可用、以及与企业现有系统的集成。

这个项目采用了微服务网关架构，通过 ZooKeeper 做服务注册与发现，将 RAG 服务、PPT Agent 服务、GPT Workflow 服务等多个模块统一在 API Gateway 后面，前端使用 Vue 3 + Vite，后端各服务独立部署、独立扩展。

技术选型

组件	选择	理由
网关	FastAPI（微服务网关）	请求路由、认证聚合、服务发现
注册中心	ZooKeeper	成熟稳定，社区支持好
RAG 服务	Python FastAPI + LangChain	与 BigGraph 能力复用
PPT Agent	Python LangGraph	多步推理生成 PPT
工作流引擎	GPT Workflow	可视化任务编排
前端	Vue 3 + Vite	现代前端框架，构建速度快
容器化	Docker / Docker Compose	一键部署，环境一致
数据库	PostgreSQL + Qdrant + Redis	混合存储方案

架构设计

整体采用"API Gateway + 微服务"架构：

Client → API Gateway (FastAPI) → ZooKeeper 注册中心
  ├── /api/rag/* → RAG Service
  ├── /api/ppt/* → PPT Agent Service
  └── /api/workflow/* → GPT Workflow Service
``\

**网关层承担四大职责：**
1. **统一入口**：所有客户端请求统一经过网关
2. **认证鉴权**：在网关层统一做 JWT 认证和权限校验
3. **请求路由**：根据路径前缀转发到对应微服务
4. **服务发现**：通过 ZooKeeper 动态获取服务实例地址

## 核心实现

### 微服务网关

```python
def create_app() -> FastAPI:
    app = FastAPI(title="Enterprise AI Microservice Gateway", lifespan=lifespan)
    app.add_middleware(CORSMiddleware, allow_origins=["*"],
                       allow_credentials=True, allow_methods=["*"],
                       allow_headers=["*"])

    @app.get("/health")
    def health():
        return {"status": "ok", "service": settings.gateway_name}

    @app.get("/api/services")
    def list_services():
        registry = getattr(app.state, "registry", ZookeeperServiceRegistry())
        return {"services": [s.to_public_dict() for s in registry.list_services()]}

    # 路由分发
    @app.api_route("/api/rag/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
    async def rag_proxy(path: str, request: Request):
        registry = getattr(app.state, "registry", ZookeeperServiceRegistry())
        return await proxy_request(request, registry, "rag-service", f"/api/{path}")

    @app.api_route("/api/ppt/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
    async def ppt_proxy(path: str, request: Request):
        registry = getattr(app.state, "registry", ZookeeperServiceRegistry())
        return await proxy_request(request, registry, "ppt-agent-service", f"/api/{path}")

    @app.api_route("/api/workflow/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
    async def workflow_proxy(path: str, request: Request):
        registry = getattr(app.state, "registry", ZookeeperServiceRegistry())
        return await proxy_request(request, registry, "gpt-workflow-service", f"/api/{path}")

    return app

ZooKeeper 服务注册

class ZookeeperServiceRegistry:
    def __init__(self, hosts: str = "localhost:2181"):
        self.client = KazooClient(hosts=hosts)
        self.base_path = "/services"

    def start(self):
        self.client.start()
        self.client.ensure_path(self.base_path)

    def register(self, service_name: str, instance_id: str, metadata: dict):
        path = f"{self.base_path}/{service_name}/{instance_id}"
        # 创建临时节点，服务下线自动删除
        self.client.create(path, json.dumps(metadata).encode(),
                          ephemeral=True, sequence=False)

    def list_services(self) -> list[ServiceInstance]:
        services = []
        for service_name in self.client.get_children(self.base_path):
            service_path = f"{self.base_path}/{service_name}"
            for instance_id in self.client.get_children(service_path):
                data, _ = self.client.get(f"{service_path}/{instance_id}")
                services.append(ServiceInstance.from_dict(json.loads(data)))
        return services

用户认证与注册

@app.post("/api/auth/register", response_model=AuthResponse)
def register(payload: RegisterRequest, store: AuthStore = Depends(get_auth_store)):
    return store.register(payload)

@app.post("/api/auth/login", response_model=AuthResponse)
def login(payload: LoginRequest, store: AuthStore = Depends(get_auth_store)):
    return store.login(payload)

@app.get("/api/auth/me", response_model=UserPublic)
def me(current_user: UserPublic = Depends(require_current_user)):
    return current_user

踩坑记录

1. ZooKeeper 会话超时
初期 ZooKeeper 会话超时时间设置过短（5秒），微服务在网络波动时会频繁断开重连，导致注册信息抖动。解决方案：将 session_timeout 调整为 30 秒，并加入重连监听器，在重连后自动重新注册。

2. 网关级联超时
当某个微服务响应慢时，网关线程被阻塞，级联影响到其他服务的请求。解决方案：在网关层设置 HTTP 客户端超时（连接超时 5s，读取超时 30s），并为每个上游服务配置独立的连接池。

3. 前端跨域问题
Vue 前端开发时通过 Vite 代理，但生产环境通过 Nginx 反向代理遇到了 CORS 问题。解决方案：在网关层统一配置 CORS 中间件（允许所有来源），前端生产环境通过同域 Nginx 访问。

总结

Enterprise_RAG_Agent_Platform 教会了我"企业级"三个字的真正含义。技术选型上，不一定要用最前沿的技术，但一定要用最成熟的方案。ZooKeeper 虽然"老"，但在服务发现场景下非常稳定。微服务网关统一了认证、路由和服务发现，让后续新增微服务的成本极低。

这个项目的核心价值在于证明了 RAG + Agent 在企业场景下的可行性——通过微服务架构实现模块解耦，通过容器化实现标准化部署，通过统一的网关实现安全管控。它是 BigGraph 从"能做"到"能用"的关键一步。