CrewAI 生产化：缓存、回调、LLM 配置

概念速查

CrewAI 在生产环境中需要关注三个核心配置维度。Cache 机制对 Tool 调用做输入哈希缓存——相同输入命中缓存则跳过执行，直接返回上次结果。Callbacks是挂载在 Crew 事件上的钩子函数，用于日志、监控和远程追踪。LLM 配置在 Agent 级别通过 llm 参数传入模型对象，支持 model、temperature、max_tokens、api_key 等参数的不同组合。

# crewai>=0.30.0
from crewai import Agent, Crew
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,
    max_tokens=4096,
    api_key="sk-xxx",
)

cache_llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    max_tokens=512,
)

writer = Agent(
    role="Writer",
    goal="Write content",
    backstory="You are a writer.",
    llm=llm,
    cache=True,
)

底层原理

Cache：输入哈希与跨 Agent 共享

每次 Tool 调用时，CrewAI 对工具名称和参数做哈希处理，以哈希值为键查本地内存缓存。命中缓存则直接返回上次结果，不走模型调用。缓存在 Crew 级别共享——同一 Crew 内不同 Agent 调用相同工具和参数时也能命中。生产环境建议始终开启缓存，其内存开销极低（KB 级别），但对重复性工具调用（如反复查询天气或数据库）的收益非常明显。

# crewai>=0.30.0
# crewai.cache.cache 内部逻辑示意
from hashlib import sha256


def _cache_key(tool_name: str, args: dict) -> str:
    raw = f"{tool_name}:{sorted(args.items())}"
    return sha256(raw.encode()).hexdigest()

Crew 构造时传 cache=True（默认开启）使该 Crew 内所有 Agent 共享同一 Cache 实例。生产环境建议始终开启，内存开销低（~KB 级），对重复工具调用（如天气查询、数据库查询）收益明显。如需跳过单次缓存，Agent 构造传 cache=False。

Callbacks：事件驱动的监控层

Crew 实例对外暴露了多种事件挂钩，包括 task 开始和完成、step 执行、tool 调用等。通过这些回调函数，可以在不修改 Agent 和 Task 定义的前提下实现日志记录、性能监控和远程追踪。本地调试时用简单 print，生产环境应接入外部监控系统。

# crewai>=0.30.0
from crewai import Crew


def on_task_started(task):
    print(f"[CREW] 开始执行: {task.description[:50]}")


def on_task_completed(task):
    print(f"[CREW] 完成: {task.description[:50]}")


crew = Crew(
    agents=[writer],
    tasks=[...],
    cache=True,
    task_callback_started=on_task_started,
    task_callback_completed=on_task_completed,
    step_callback=None,       # 每步 Agent 执行回调
    tool_callback=None,       # 工具调用回调
)

LLM 配置：模型级的灵活编排

每个 Agent 独立持有 LLM 实例，LLM 参数（model、temperature、api_key）完全由调用方控制。这是多模型混合编排的基础——分析 Agent 用强模型，分类 Agent 用小模型。

# crewai>=0.30.0
from crewai import Agent

researcher = Agent(
    role="Researcher",
    goal="Research topic",
    backstory="You are a researcher.",
    llm=ChatOpenAI(model="gpt-4o", temperature=0.3),
    cache=True,
)
reviewer = Agent(
    role="Reviewer",
    goal="Review output",
    backstory="You are a reviewer.",
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    cache=True,
)

架构设计原则

原则 1：缓存策略按场景区分

Crew 级别 cache=True 是生产默认值。如果 Pipeline 涉及多次调用同一 Tool（如爬虫反复查询价格或数据库），缓存显著降低成本和延迟。但对于对实时性要求高的场景——比如汇率查询或股票价格——缓存会导致 Agent 拿到过期数据。需要根据工具的语义特性来决定缓存开关，而非一刀切全场开启。建议对纯查询类工具保持缓存，对有时效性要求的工具通过自定义 cache_function 做时间窗口判断。

原则 2：回调函数轻量化

本地调试可以用 task_callback_started 和 task_callback_completed 直接打印日志；但生产环境下不应当在回调中执行阻塞 I/O 操作。回调函数的执行会延长 Agent 的响应时间，阻塞 I/O 会使整个 Crew 的执行变慢。正确的做法是回调中只做入队操作或者异步写入，将耗时操作交给后台消费者处理。

缓存策略：Crew 级别 cache=True 是生产默认值。如果 Pipeline 涉及多次调用同一 Tool（如多次查价格），缓存显著降本。仅在工具结果具有时效性时（如实时汇率）关闭。

回调配置：本地调试用 task_callback_started / task_callback_completed 打日志；远程追踪接入 LangSmith 或自建监控时将回调指向 SDK 上报函数。不要在同个回调里做阻塞 I/O + 业务逻辑，拆成两个回调或异步队列。

# crewai>=0.30.0 — 本地调试 vs 远程追踪
from langsmith import Client

client = Client(api_key="ls-xxx")


def trace_to_langsmith(task):
    client.create_run(
        name=task.description[:60],
        inputs={"description": task.description},
        run_type="chain",
    )


crew = Crew(
    agents=[writer, reviewer],
    tasks=[...],
    cache=True,
    task_callback_started=trace_to_langsmith,
)

LLM 配置分层：Agent 定义时不写死 LLM，通过环境变量或配置中心注入。api_key 仅从环境变量读取，禁止硬编码。temperature 对确定性任务（分类、提取）设 0，对创编任务设 0.7—0.9。

性能建议：

• 缓存始终开启，只对时效敏感任务按需关闭
• 回调函数轻量化，耗时操作异步出队
• 多 Agent 间按职责分配不同模型，避免全用最大模型
• max_tokens 根据任务输出量设定，不给 Agent 留无上限的 token 窗口