垦荒学园
  • 中文

      • Memory 记忆

      CrewAI 提供统一的 Memory 类，不再拆成互不相干的短期 / 长期 / 实体 / 外部多种类型。保存时由 LLM 推断 scope、类别与重要性；召回按语义相似度、新近度、重要性的综合分排序。

      四种用法：独立脚本、Crew、Agent、Flow 内部。

      独立使用

      from crewai import Memory

memory = Memory()
memory.remember("We decided to use PostgreSQL for the user database.")
matches = memory.recall("What database did we choose?")
for m in matches:
    print(f"[{m.score:.2f}] {m.record.content}")

memory.forget(scope="/project/old")
print(memory.tree())

      可调 recency_weight、semantic_weight、importance_weight、recency_half_life_days。extract_memories(长文本) 会拆成原子事实，再逐条 remember。

      未指定 scope 时，LLM 根据内容把记忆挂到类似文件系统的路径上（/project/decisions、/agent/researcher 等）。recall 在当前分支内搜索，更准也更快。也可 remember(..., scope="/research/databases") 显式指定。

      与 Crew

      from crewai import Crew, Process, Memory

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    memory=True,
    verbose=True,
)

memory = Memory(recency_weight=0.4, semantic_weight=0.4, importance_weight=0.2)
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    memory=memory,
)

      memory=True 会创建默认 Memory()，并沿用 Crew 的 embedder。未自定义 embedder 时，记忆侧默认 OpenAI text-embedding-3-large（需要可用的 Embedding API）。任务结束后自动抽取事实写入；任务开始前召回并注入提示。Crew 内 Agent 默认共享这份记忆。

      与 Agent（私有 scope）

      from crewai import Agent, Memory

memory = Memory()
researcher = Agent(
    role="Researcher",
    goal="Find and analyze information",
    backstory="Expert researcher with attention to detail",
    memory=memory.scope("/agent/researcher"),
)

      研究员只看见 /agent/researcher 分支；写作者若不设 memory，则使用 Crew 共享记忆。适合「草稿私有、成稿公开」。

      与 Flow

      每个 Flow 自带记忆 API：self.remember()、self.recall()、self.extract_memories()。

      from crewai.flow.flow import Flow, listen, start

class ResearchFlow(Flow):
    @start()
    def gather_data(self):
        findings = "PostgreSQL handles 10k concurrent connections."
        self.remember(findings, scope="/research/databases")
        return findings

    @listen(gather_data)
    def write_report(self, findings):
        past = self.recall("database performance benchmarks")
        context = "\n".join(f"- {m.record.content}" for m in past)
        return f"Report:\nNew findings: {findings}\nPrevious context:\n{context}"

      把跨步骤要复用的事实放进 Memory；把本轮编排变量（当前 topic、是否通过质检）放在 Flow state。两者职责不同：state 是工作流变量，Memory 是可检索知识。

      与 LangChain 检查点

      LangChain 记忆 强调 thread_id + checkpointer（会话消息持久化）。CrewAI Memory 强调 跨任务的事实记忆与 scope 树。长流程断点续跑用 Flow 的 @persist 或 Crew checkpoint，不要只用 remember 代替状态机。

      Embedding 成本、隐私与密钥管理与任何 RAG 系统相同：见 RAG。不要把密钥或用户 PII 无过滤地 remember。

      下一步

      评论