本文复盘了我们团队在构建一套面向中小企业的多Agent智能办公平台时的完整技术决策过程。文章从单体LLM调用的局限性出发，详细阐述了基于FastAPI + Celery + Milvus的分布式Agent编排架构设计，重点拆解了插件化技能市场、RAG专家推理管道、Cron定时调度引擎以及多租户RBAC资源隔离的工程实现细节。文中包含了大量核心模块的代码示例与生产环境踩坑记录，期望能为同类系统的架构设计提供参考。

一、背景：为什么单体LLM撑不起企业办公场景 

我们团队在早期版本中采用了一个看似简单的架构：前端接收用户输入，后端拼接Prompt，直接调用云端大模型的Chat Completion API，返回结果。这套方案在Demo阶段表现良好，但一旦投入真实办公场景，立刻暴露出四个无法回避的问题：

通用大模型对垂直领域（如3D打印工艺参数、银行信贷合规条款、中药材鉴别标准）缺乏深度认知。我们做过一组对比测试：在法律合同审查任务中，GPT-4的条文引用错误率达到23%，而在我们的"法律专家Agent"引入RAG后，这一数字降至4%以下。

2. 长上下文窗口的边际效益递减

当我们将市场调研所需的竞品数据、行业报告、政策文件一次性塞入Prompt时，token消耗呈线性增长，但输出质量在超过12k tokens后出现明显衰减。这不是模型能力问题，而是Attention机制在超长上下文中对远距离信息的关注度自然下降。

在"生成商品宣传视频"这类复合任务中，需要依次完成：需求解析 → 脚本生成 → 分镜规划 → 素材匹配 → 文案输出。如果把所有指令写在一个Prompt里，任何一个环节出错都会导致整条链路失败，且难以定位问题节点。

当多个团队成员同时使用系统时，长耗时的研报生成任务会占用大量连接池资源，导致简单的问答请求排队等待。我们需要一种机制来实现任务优先级调度和资源配额隔离。

基于以上痛点，我们决定重构系统，引入

。每个Agent承担一个具体角色（如法律专家、市场分析师、海报设计师），通过编排层进行任务分发和结果聚合。这套系统目前已稳定支撑数百家中小企业的日常办公需求。

二、总体架构：分层解耦与异步通信 

┌─────────────────────────────────────────────┐
│              接入层 (FastAPI Gateway)         │
│  鉴权(JWT) · 限流(Redis) · 请求路由          │
├─────────────────────────────────────────────┤
│           编排层 (Agent Orchestrator)         │
│  NLU意图识别 · Agent匹配 · 任务生命周期管理    │
├─────────────────────────────────────────────┤
│           执行层 (Agent Pool + Skills)        │
│  专家Agent(Docker沙箱) · 技能插件(Python包)   │
├─────────────────────────────────────────────┤
│         基础设施层 (Data & Message)           │
│  PostgreSQL · Milvus · Redis(Celery Broker)  │
└─────────────────────────────────────────────┘

2.1 为什么选择Celery而非WebSocket长连接 

初期我们试图用WebSocket维持前端与Agent之间的长连接，但在以下场景中出现问题：

Celery异步任务 + 前端轮询/Webhook回调

# gateway/router.py
from fastapi import FastAPI, HTTPException, Depends
from celery_app import agent_task_queue
import asyncio
import redis.asyncio as redis

app = FastAPI(title="Multi-Agent Office Platform")
redis_client = redis.Redis(host="localhost", port=6379, db=0)

class TaskStatus:
    PENDING = "PENDING"
    RUNNING = "RUNNING"
    SUCCESS = "SUCCESS"
    FAILED = "FAILED"

@app.post("/api/v1/chat")
async def dispatch_chat(payload: dict, user: dict = Depends(verify_jwt)):
    """
    接收用户请求，异步分发至Agent执行
    """
    tenant_id = user["tenant_id"]
    
    # 前置配额检查：该租户的并发任务数是否超限
    concurrency_key = f"concurrency:{tenant_id}"
    current = await redis_client.incr(concurrency_key)
    await redis_client.expire(concurrency_key, 3600)
    
    if current > user["qps_limit"]:
        await redis_client.decr(concurrency_key)
        raise HTTPException(status_code=429, detail="Concurrency quota exceeded")
    
    # 提交异步任务
    task = agent_task_queue.delay(
        tenant_id=tenant_id,
        user_id=user["user_id"],
        message=payload["message"],
        context=payload.get("context", {})
    )
    
    # 写入Redis状态
    await redis_client.hset(
        f"task:{task.id}",
        mapping={
            "status": TaskStatus.PENDING,
            "tenant_id": tenant_id,
            "created_at": asyncio.get_event_loop().time()
        }
    )
    
    return {"task_id": task.id, "status": TaskStatus.PENDING}

编排层的核心是NLU引擎，负责将用户的自然语言输入映射到具体的Agent。我们没有选择微调模型来做意图分类，而是采用

# orchestrator/nlu_router.py
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

class NLURouter:
    def __init__(self):
        self.encoder = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
        # 预定义的Agent描述向量库
        self.agent_descriptions = {
            "legal_expert": "合同审查 法律问答 合规风险 劳动法 知识产权",
            "market_analyst": "市场调研 竞品分析 行业报告 消费者画像",
            "designer": "海报设计 宣传图 图片生成 视觉素材",
            "finance_expert": "财务报表 税务筹划 融资并购 预算管理",
            # ... 更多Agent
        }
        self.agent_embeddings = {
            name: self.encoder.encode(desc)
            for name, desc in self.agent_descriptions.items()
        }
    
    def route(self, user_input: str, threshold: float = 0.65) -> tuple[str, float]:
        """
        返回匹配的Agent名称和置信度
        """
        input_vec = self.encoder.encode(user_input).reshape(1, -1)
        scores = {}
        for name, emb in self.agent_embeddings.items():
            scores[name] = cosine_similarity(input_vec, emb.reshape(1, -1))[0][0]
        
        best_match = max(scores, key=scores.get)
        confidence = scores[best_match]
        
        if confidence < threshold:
            return "general_assistant", confidence
        return best_match, confidence

（向量编码在本地CPU上耗时<50ms）、

（新增Agent只需添加描述文本）。在生产环境中，意图匹配准确率达到89%，剩余11%由用户手动从推荐列表中选择纠正。

三、技能市场：插件化架构的工程实现 

系统的扩展性很大程度上取决于技能市场的设计。我们的目标是让第三方开发者能够在不修改主干代码的前提下，向平台注入新的AI能力。

# skills/base.py
from abc import ABC, abstractmethod
from pydantic import BaseModel, Field
from typing import Any, Dict, Optional

class SkillContext(BaseModel):
    tenant_id: str
    user_id: str
    params: Dict[str, Any] = Field(default_factory=dict)
    conversation_history: list = Field(default_factory=list)
    metadata: Dict[str, Any] = Field(default_factory=dict)

class SkillResult(BaseModel):
    success: bool
    data: Any = None
    error_msg: Optional[str] = None
    tokens_used: int = 0
    execution_time_ms: int = 0

class BaseSkill(ABC):
    name: str = ""
    version: str = "0.1.0"
    description: str = ""
    required_params: list = []
    
    @abstractmethod
    async def execute(self, ctx: SkillContext) -> SkillResult:
        """技能执行入口，必须由子类实现"""
        pass
    
    def validate_params(self, ctx: SkillContext) -> Optional[str]:
        """参数校验，返回错误信息或None"""
        missing = [p for p in self.required_params if p not in ctx.params]
        if missing:
            return f"Missing required parameters: {missing}"
        return None

技能安装后存储在独立目录中，通过Python的

动态加载。为防止恶意代码或内存泄漏影响主干服务，每个技能的执行被限制在独立的Docker容器中：

# skills/loader.py
import importlib.util
import subprocess
import tempfile
import os

class SkillLoader:
    SKILL_DIR = "/opt/skills/"
    
    @classmethod
    def load_from_disk(cls, skill_name: str) -> BaseSkill:
        """从磁盘动态加载技能模块"""
        spec = importlib.util.spec_from_file_location(
            skill_name,
            os.path.join(cls.SKILL_DIR, skill_name, "main.py")
        )
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        
        if not hasattr(module, "Skill") or not issubclass(module.Skill, BaseSkill):
            raise ImportError(f"Skill '{skill_name}' does not implement BaseSkill")
        
        instance = module.Skill()
        if not instance.name:
            instance.name = skill_name
        return instance
    
    @classmethod
    async def execute_in_sandbox(cls, skill_name: str, ctx: SkillContext) -> SkillResult:
        """
        在Docker沙箱中执行技能，通过临时文件传递上下文和结果
        """
        with tempfile.NamedTemporaryFile(mode='w+', suffix='.json') as input_f:
            with tempfile.NamedTemporaryFile(mode='w+', suffix='.json') as output_f:
                # 写入上下文
                input_f.write(ctx.json())
                input_f.flush()
                
                # 启动沙箱容器
                cmd = [
                    "docker", "run", "--rm",
                    "--memory=512m",           # 内存限制
                    "--cpus=1.0",              # CPU限制
                    "--network=none",          # 禁止网络访问（除非技能声明需要）
                    "-v", f"{input_f.name}:/input.json:ro",
                    "-v", f"{output_f.name}:/output.json",
                    f"skill-{skill_name}:latest",
                    "python", "/app/run.py", "/input.json", "/output.json"
                ]
                
                proc = subprocess.run(cmd, capture_output=True, timeout=120)
                if proc.returncode != 0:
                    return SkillResult(
                        success=False,
                        error_msg=f"Sandbox execution failed: {proc.stderr.decode()}"
                    )
                
                # 读取结果
                output_f.seek(0)
                return SkillResult.parse_raw(output_f.read())

这套沙箱机制的额外开销约为200-400ms（容器启动时间），对于耗时通常在数秒以上的复杂技能来说完全可以接受。

四、专家Agent：RAG推理管道的深度优化 

专家库是平台的核心竞争力。与通用聊天机器人不同，专家Agent需要在特定领域提供

的专业回答。我们的技术方案是

以"银行信贷专家"为例，我们需要将数千份监管文件、内部操作手册、历史案例纳入知识库：

# experts/document_pipeline.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Milvus
import hashlib

class DocumentPipeline:
    def __init__(self):
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=64,          # 重叠窗口，防止语义断裂
            length_function=len,
            separators=["\n\n", "\n", "。", "！", "？", " ", ""]
        )
        self.embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
        )
        self.vectorstore = Milvus(
            embedding_function=self.embeddings,
            collection_name="bank_expert_knowledge",
            connection_args={"host": "milvus.internal", "port": "19530"}
        )
    
    def ingest(self, doc_path: str, metadata: dict):
        """文档入库流程"""
        with open(doc_path, 'r', encoding='utf-8') as f:
            text = f.read()
        
        chunks = self.text_splitter.split_text(text)
        
        # 为每个chunk生成唯一ID，用于去重
        metadatas = []
        ids = []
        for i, chunk in enumerate(chunks):
            chunk_hash = hashlib.md5(f"{doc_path}_{i}_{chunk}".encode()).hexdigest()
            ids.append(chunk_hash)
            meta = metadata.copy()
            meta.update({"chunk_index": i, "source": doc_path})
            metadatas.append(meta)
        
        self.vectorstore.add_texts(
            texts=chunks,
            metadatas=metadatas,
            ids=ids
        )

单纯依赖向量相似度的检索存在一个问题：

。例如用户问"《商业银行法》第36条对担保的规定"，向量检索可能因为"商业银行法"和"担保"的语义距离较远而漏掉最相关的法条。

BM25稀疏检索 + 向量稠密检索的加权融合

# experts/hybrid_retriever.py
from rank_bm25 import BM25Okapi
import numpy as np
from typing import List, Tuple

class HybridRetriever:
    def __init__(self, vectorstore, documents: List[str], metadatas: List[dict]):
        self.vectorstore = vectorstore
        self.documents = documents
        self.metadatas = metadatas
        
        # BM25索引
        tokenized_docs = [self._tokenize(d) for d in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
    
    def _tokenize(self, text: str) -> List[str]:
        """中文分词（简化版，实际用jieba）"""
        import jieba
        return list(jieba.cut(text))
    
    def retrieve(self, query: str, top_k: int = 5, alpha: float = 0.7) -> List[dict]:
        """
        alpha: 向量检索权重 (1-alpha): BM25权重
        """
        # 向量检索得分
        vector_results = self.vectorstore.similarity_search_with_score(query, k=top_k*2)
        vector_scores = {r[0].page_content: r[1] for r in vector_results}
        
        # BM25检索得分
        tokenized_query = self._tokenize(query)
        bm25_scores = self.bm25.get_scores(tokenized_query)
        
        # 归一化并加权融合
        all_docs = list(vector_scores.keys()) + self.documents[:len(bm25_scores)]
        combined_scores = {}
        
        for i, doc in enumerate(all_docs):
            vs = vector_scores.get(doc, 0)
            bs = bm25_scores[i] if i < len(bm25_scores) else 0
            
            # Min-Max归一化
            vs_norm = self._normalize(vs, list(vector_scores.values()))
            bs_norm = self._normalize(bs, bm25_scores.tolist())
            
            combined_scores[doc] = alpha * vs_norm + (1 - alpha) * bs_norm
        
        # 排序取Top-K
        sorted_docs = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
        
        return [
            {"content": doc, "score": score, "metadata": self._get_metadata(doc)}
            for doc, score in sorted_docs
        ]
    
    def _normalize(self, val, all_vals):
        min_v, max_v = min(all_vals), max(all_vals)
        if max_v - min_v == 0:
            return 0
        return (val - min_v) / (max_v - min_v)

在实际测试中，混合检索的Recall@5比纯向量检索提升了18%，尤其在法律法规条文、技术参数表等

检索到上下文后，专家Agent的推理流程如下：

# experts/expert_agent.py
class ExpertAgent:
    def __init__(self, domain: str, retriever: HybridRetriever):
        self.domain = domain
        self.retriever = retriever
        self.llm_client = OpenAIClient()  # 封装好的LLM调用客户端
    
    async def consult(self, query: str, conversation_id: str = None) -> str:
        # Step 1: 检索相关知识
        contexts = self.retriever.retrieve(query, top_k=5)
        
        # Step 2: 构建System Prompt
        system_prompt = self._build_system_prompt(contexts)
        
        # Step 3: 获取历史对话（如果有）
        history = []
        if conversation_id:
            history = ConversationStore.get(conversation_id)
        
        # Step 4: 流式生成
        response_stream = self.llm_client.chat_completion(
            messages=[
                {"role": "system", "content": system_prompt},
                *history,
                {"role": "user", "content": query}
            ],
            stream=True,
            temperature=0.3,      # 低温度保证专业性
            max_tokens=2048
        )
        
        # Step 5: 收集完整回复并存储
        full_response = ""
        async for chunk in response_stream:
            content = chunk.choices[0].delta.content
            if content:
                full_response += content
        
        if conversation_id:
            ConversationStore.append(conversation_id, {
                "user": query,
                "assistant": full_response
            })
        
        return full_response
    
    def _build_system_prompt(self, contexts: List[dict]) -> str:
        context_str = "\n\n---\n\n".join([
            f"[来源: {c['metadata'].get('source', '未知')}]\n{c['content']}"
            for c in contexts
        ])
        
        return f"""你是一位资深的{self.domain}专家。请基于以下参考材料回答用户的问题。
要求：
1. 回答必须严格基于参考材料，不得编造材料中不存在的信息
2. 每条关键结论后标注来源（如：[来源: xxx.pdf]）
3. 如果参考材料中信息不足以回答问题，明确告知用户"根据现有资料无法回答"，不要猜测
4. 使用专业但易懂的语言

参考材料：
{context_str}
"""

在System Prompt中强制要求标注来源

。这不仅提升了答案的可信度，也让用户能够自行核实信息的准确性。

五、定时任务引擎：从被动响应到主动执行 

企业办公场景中，很多任务需要周期性自动执行——例如每日监控竞品动态并生成研报、每周汇总行业政策变化、每月生成财务分析报告。我们将这类需求抽象为

，技术实现基于Celery Beat + Redis。

# scheduler/models.py
from pydantic import BaseModel
from typing import Optional
from enum import Enum

class TaskStatus(str, Enum):
    ACTIVE = "active"
    PAUSED = "paused"
    COMPLETED = "completed"

class ScheduleConfig(BaseModel):
    cron_expression: str       # 标准5段Cron: "0 15 * * *"
    timezone: str = "Asia/Shanghai"
    max_retries: int = 3
    timeout_seconds: int = 300

class ScheduledTask(BaseModel):
    id: str
    tenant_id: str
    name: str
    description: str
    agent_name: str
    prompt_template: str       # 支持变量替换，如 "分析{company}的最新动态"
    schedule: ScheduleConfig
    status: TaskStatus = TaskStatus.ACTIVE
    last_run_at: Optional[int] = None
    next_run_at: Optional[int] = None
    created_by: str

# scheduler/engine.py
from celery import Celery
from croniter import croniter
import time
import pytz
from datetime import datetime

celery_app = Celery("scheduler", broker="redis://localhost:6379/1")

class SchedulerEngine:
    def __init__(self):
        self.tasks: dict = {}  # task_id -> ScheduledTask
    
    def register_task(self, task: ScheduledTask):
        """注册任务到调度器"""
        self.tasks[task.id] = task
        self._calculate_next_run(task)
        self._register_celery_periodic(task)
    
    def _register_celery_periodic(self, task: ScheduledTask):
        """转换为Celery Beat的periodic task"""
        minute, hour, dom, month, dow = task.schedule.cron_expression.split()
        
        celery_app.conf.beat_schedule[f"task_{task.id}"] = {
            "task": "scheduler.tasks.execute_scheduled_task",
            "schedule": crontab(
                minute=minute,
                hour=hour,
                day_of_month=dom,
                month_of_year=month,
                day_of_week=dow
            ),
            "args": (task.id,),
            "options": {
                "expires": task.schedule.timeout_seconds
            }
        }
    
    def _calculate_next_run(self, task: ScheduledTask):
        """计算下次执行时间"""
        tz = pytz.timezone(task.schedule.timezone)
        now = datetime.now(tz)
        cron = croniter(task.schedule.cron_expression, now)
        task.next_run_at = int(cron.get_next(float))

# Celery任务执行函数
@celery_app.task(bind=True, max_retries=3)
def execute_scheduled_task(self, task_id: str):
    engine = SchedulerEngine()
    task = engine.tasks.get(task_id)
    if not task or task.status != TaskStatus.ACTIVE:
        return
    
    try:
        # 实例化目标Agent
        agent = AgentFactory.get(task.agent_name)
        
        # 渲染Prompt模板（支持变量替换）
        prompt = render_template(task.prompt_template, {
            "date": datetime.now().strftime("%Y-%m-%d"),
            "tenant_id": task.tenant_id
        })
        
        # 执行任务
        result = agent.run(prompt, timeout=task.schedule.timeout_seconds)
        
        # 持久化结果
        ResultStore.save(task_id, {
            "output": result,
            "executed_at": int(time.time()),
            "status": "success"
        })
        
        # 发送通知（Webhook/邮件/站内信）
        NotificationService.notify(
            tenant_id=task.tenant_id,
            title=f"定时任务「{task.name}」执行完成",
            content=result[:500] + "..."
        )
        
        # 更新last_run_at并计算next_run_at
        task.last_run_at = int(time.time())
        engine._calculate_next_run(task)
        
    except Exception as exc:
        # 失败重试，指数退避
        retry_count = self.request.retries
        if retry_count < task.schedule.max_retries:
            raise self.retry(exc=exc, countdown=60 * (2 ** retry_count))
        
        # 达到最大重试次数，标记失败
        ResultStore.save(task_id, {
            "error": str(exc),
            "executed_at": int(time.time()),
            "status": "failed"
        })

在K8s部署环境下，Beat进程有多个副本时，同一任务会被多次触发。解决方案是使用

@celery_app.task(bind=True)
def execute_scheduled_task(self, task_id: str):
    lock_key = f"lock:task:{task_id}"
    lock = redis_client.lock(lock_key, timeout=300)
    
    if not lock.acquire(blocking=False):
        logger.info(f"Task {task_id} is already running, skipping")
        return
    
    try:
        # 执行任务逻辑...
        pass
    finally:
        lock.release()

坑2：Cron时区不一致导致执行时间偏移

Docker容器的默认时区是UTC，而我们配置的Cron是基于Asia/Shanghai。结果是任务总是在预期时间+8小时才执行。解决方案是在Celery配置中显式指定时区：

celery_app.conf.timezone = "Asia/Shanghai"
celery_app.conf.enable_utc = False

六、多租户RBAC与资源配额管理 

企业级系统必须解决多租户隔离问题。我们的方案基于

RBAC（基于角色的访问控制）+ Redis信号量

-- tenants 表：企业租户
CREATE TABLE tenants (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    total_seats INT NOT NULL DEFAULT 10,
    used_seats INT NOT NULL DEFAULT 0,
    qps_limit INT NOT NULL DEFAULT 5,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- users 表：成员账号
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID REFERENCES tenants(id),
    phone VARCHAR(20) UNIQUE NOT NULL,
    role VARCHAR(50) NOT NULL DEFAULT 'member',
    seat_assigned BOOLEAN NOT NULL DEFAULT FALSE,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 角色枚举：super_admin | admin | member | viewer

# middleware/quota.py
from fastapi import Request, HTTPException
import redis.asyncio as redis

class QuotaMiddleware:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    async def __call__(self, request: Request, call_next):
        user = request.state.user  # 由鉴权中间件注入
        if not user:
            return await call_next(request)
        
        tenant_id = user["tenant_id"]
        
        # 并发数控制（滑动窗口）
        window_key = f"concurrency:{tenant_id}"
        current = await self.redis.incr(window_key)
        if current == 1:
            await self.redis.expire(window_key, 60)  # 60秒窗口
        
        tenant_qps = await self._get_tenant_qps(tenant_id)
        if current > tenant_qps:
            await self.redis.decr(window_key)
            raise HTTPException(
                status_code=429,
                detail={"error": "QuotaExceeded", "retry_after": 30}
            )
        
        try:
            response = await call_next(request)
            return response
        finally:
            await self.redis.decr(window_key)
    
    async def _get_tenant_qps(self, tenant_id: str) -> int:
        # 从缓存或DB获取租户的QPS限制
        cached = await self.redis.get(f"tenant_qps:{tenant_id}")
        if cached:
            return int(cached)
        # fallback到DB查询
        # ...
        return 5  # default

坐席（Seat）是平台的核心计费单元，分配操作必须是原子性的：

# services/seat_service.py
from sqlalchemy.orm import Session
from sqlalchemy import select, update

class SeatService:
    def __init__(self, db: Session):
        self.db = db
    
    def assign_seat(self, tenant_id: str, user_id: str) -> bool:
        """
        为成员分配坐席，使用行级锁防止超卖
        """
        with self.db.begin():
            # SELECT FOR UPDATE 锁定租户行
            tenant = self.db.execute(
                select(Tenant).where(Tenant.id == tenant_id).with_for_update()
            ).scalar_one()
            
            if tenant.used_seats >= tenant.total_seats:
                return False
            
            # 更新用户坐席状态
            user = self.db.execute(
                select(User).where(User.id == user_id)
            ).scalar_one()
            
            if user.seat_assigned:
                return True  # 已经分配过了
            
            user.seat_assigned = True
            tenant.used_seats += 1
            self.db.commit()
            return True

回顾整个系统的构建过程，几个关键的技术决策值得总结：

未来我们计划在三個方向上继续迭代：一是引入

，让多个专家Agent能够自主协作完成复杂任务（如"法律专家+财务专家"联合进行并购尽职调查）；二是探索

，降低对云端大模型API的依赖；三是构建

，根据用户反馈自动优化RAG检索策略和Prompt模板。

犀牛卫企业级多Agent协同平台的架构设计与工程实践

本文复盘了我们团队在构建一套面向中小企业的多Agent智能办公平台时的完整技术决策过程。文章从单体LLM调用的局限性出发，详细阐述了基于FastAPI + Celery + Milvus的分布式Agent编排架构设计，重点拆解了插件化技能市场、RAG专家推理管道、Cron定时调度引擎以及多租户RBAC资源隔离的工程实现细节。文中包含了大量核心模块的代码示例与生产环境踩坑记录，期望能为同类系统的架构

本文详细解析了多Agent智能办公平台的技术架构，涵盖分布式Agent编排、插件化技能市场、RAG专家系统、定时任务引擎等核心模块。重点解决企业办公场景中的LLM局限性问题，提供可落地的异步通信、混合检索、沙箱隔离等解决方案，为中小企业AI办公系统开发提供实践参考。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

犀牛卫企业级多Agent协同平台的架构设计与工程实践

犀牛卫企业级多Agent协同平台的架构设计与工程实践

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐