5.4 部署策略

5.4 部署策略

本节约处在第五章第四站。能跑的 Crew 在笔记本上不等于能服务业务。部署形态决定了它的吞吐、可观测性和故障面。我们把常见形态按"交互方式"分成三类，配上取舍，帮你按场景选。

先把三种部署形态画成对照，看清各

先把三种部署形态画成对照，看清各自适合什么触发方式：

形态一：脚本或定时任务。适合"每天跑一次行业简报""每小时聚一批数据"这类周期批处理。写好 kickoff 封装，用系统 cron 或调度器触发，跑完把结果推到下游（数据库、邮件、看板）。优点最简单，缺点是无人在环时出错只能事后发现。

# 5.4 部署策略
import os
from crewai import Crew

def build_crew():
    # 组装研究员+写手+审核的 Crew
    return Crew(agents=[...], tasks=[...], process=Process.sequential,
                step_callback=persist_step, verbose=False)

if __name__ == "__main__":
    crew = build_crew()
    result = crew.kickoff(inputs={"topic": os.environ["TOPIC"]})
    save_to_db(str(result))   # 落库，供下游消费
​

形态二：常驻服务 API。适合"

形态二：常驻服务 API。适合"用户在前端提交主题、实时返回报告"这类在线请求。用 FastAPI 之类包一层，把 Crew 的 kickoff 暴露成接口。注意两点：Crew 本身不是线程安全的，并发请求要为每个请求新建实例或加锁；模型调用慢，接口要异步返回，别让 HTTP 连接干等。

from fastapi import FastAPI
from crewai import Crew, Process

app = FastAPI()

def make_crew(topic: str) -> Crew:
    return Crew(agents=[...], tasks=[...], process=Process.sequential, verbose=False)

@app.post("/report")
async def report(topic: str):
    # 每请求新建 Crew，避免跨请求状态串味
    crew = make_crew(topic)
    result = await crew.kickoff_async(inputs={"topic": topic})
    return {"result": str(result)}
​

形态三：事件驱动。适合"来一条消息就处理一条"的异步场景，用消息队列（如 Kafka、RabbitMQ）触发 Crew，处理完再把结果发回另一队列。好处是生产者和消费者解耦、可削峰；坏处是链路变长、需要额外的重试与死信处理。

部署时的通用纪律我们列五条：第一，模型凭证走环境变量/密钥管理，绝不进镜像；第二，限制并发与 max_rpm，防止把模型接口打爆；第三，把第四章的可观测信号接到集中日志与指标；第四，长任务加持久化与超时，崩了能续；第五，对外服务加鉴权与限流，别裸奔。

一个收尾提醒：部署形态的选择由"

一个收尾提醒：部署形态的选择由"触发方式"决定，而不是由"技术喜好"决定。先想清你的 Crew 是被 cron 调、被用户调、还是被事件调，形态自然就定。无论哪种，前面讲的记忆、人在环、可观测、持久化都不是可选装饰，而是生产运行的基本配置。下一站讲安全合规，那是部署前必须过的最后一道关。

把 Crew 送上线的几种姿势

部署不是把脚本丢服务器。按调用方式分三类：批处理（定时跑、出报告）、API 服务（被业务系统调用）、事件驱动（消息触发）。选哪种取决于"谁在什么时候要结果"。

⚠️ 常见坑：把长时间 Crew 做成同步 API，请求方超时。长任务应改成"提交任务→轮询/回调"的异步模式。另一个坑是 secrets（API Key）硬编码进代码，泄露即失控——走环境变量或密钥管理。

import os
from crewai import Agent, Task, Crew, Process

# 凭证走环境变量，绝不进代码
api_key = os.environ.get("OPENAI_API_KEY")
assert api_key, "缺少 OPENAI_API_KEY"

analyst = Agent(role="分析师", goal="出报告", backstory="严谨", verbose=False)
t = Task(description="生成日报", expected_output="日报", agent=analyst)
crew = Crew(agents=[analyst], tasks=[t], process=Process.sequential)
# 生产里：把 crew.kickoff 包成 API handler 或定时任务
print("部署配置就绪")
​

💡 关键直觉：部署是把"实验室的 Crew"变成"生产里的服务"。实验室关心能不能跑，生产关心崩了怎么办、并发怎么扛、密钥怎么守。把这三件事在部署前想清楚，上线才不慌。

# 异步提交模式：先返回任务号，再查结果
task_id = "job-1001"
status = "submitted"
assert status in ("submitted", "running", "done", "failed")
​

异步提交与状态轮询的具体实现

长任务不能做成同步阻塞接口。我们把"提交→轮询"模式落成骨架：调用方拿到 task_id 立刻返回，后台跑 Crew，调用方按 task_id 查状态，done 后取结果。这样既不超时，又能并发接多个请求。

⚠️ 常见坑：轮询间隔太短把服务打挂，太长用户体验差。我们按任务预期时长设指数间隔（1s→2s→4s），并给总超时兜底。另一个坑是 done 后结果不持久化，调用方来取时已被回收——结果必须落库再标记 done。

import time

jobs = {}  # task_id -> {"status": ..., "result": ...}

def submit(crew, inputs):
    tid = f"job-{len(jobs)+1}"
    jobs[tid] = {"status": "submitted", "result": None}
    # 真实场景放入后台线程/队列执行 crew.kickoff(inputs)
    jobs[tid]["status"] = "running"
    return tid

def poll(tid):
    return jobs.get(tid, {}).get("status", "unknown")

tid = submit("crew", {"topic": "储能"})
for _ in range(3):
    s = poll(tid)
    if s in ("done", "failed"):
        break
    time.sleep(1)
print("轮询状态:", s)
​

💡 关键直觉：把长任务当"下单"而不是"当面等"。你下单后拿小票（task_id），该干嘛干嘛，等叫号再取。这套模式让重型 Crew 也能优雅接入高并发业务系统，而不是每个请求都卡在一条同步链上。

# 幂等保护：同一 inputs 重复提交返回已有 job，避免重复烧钱
def submit_idempotent(key):
    if key in jobs:
        return key
    return submit("crew", {})

assert submit_idempotent("k1") == submit_idempotent("k1")
​