高级特性 #
记忆系统 #
记忆类型 #
text
┌─────────────────────────────────────────────────────────────┐
│ CrewAI 记忆系统 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 短期记忆（Short-term Memory） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ - 存储当前会话的上下文 │ │
│ │ - 最近交互的信息 │ │
│ │ - 临时存储，会话结束后清除 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 长期记忆（Long-term Memory） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ - 持久化的知识存储 │ │
│ │ - 跨会话的信息保留 │ │
│ │ - 使用向量数据库存储 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 实体记忆（Entity Memory） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ - 存储特定实体的信息 │ │
│ │ - 实体间的关系和属性 │ │
│ │ - 结构化的知识表示 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
启用记忆 #
python
from crewai import Agent, Crew
agent = Agent(
role="客服助手",
goal="提供个性化服务",
backstory="专业的客服代表",
memory=True
)
crew = Crew(
agents=[agent],
tasks=[task],
memory=True
)
配置 Embedder #
python
crew = Crew(
agents=[agent],
tasks=[task],
memory=True,
embedder={
"provider": "openai",
"config": {
"model": "text-embedding-3-small"
}
}
)
自定义记忆存储 #
python
from crewai.memory import LongTermMemory
crew = Crew(
agents=[agent],
tasks=[task],
memory=True,
long_term_memory=LongTermMemory(
storage=custom_storage
)
)
知识源 #
字符串知识源 #
python
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
knowledge = StringKnowledgeSource(
content="""
公司信息：
- 名称：ABC 公司
- 成立：2020 年
- 业务：AI 解决方案
"""
)
agent = Agent(
role="公司助手",
goal="回答公司相关问题",
knowledge_sources=[knowledge]
)
文件知识源 #
python
from crewai.knowledge.source.text_file_knowledge_source import TextFileKnowledgeSource
knowledge = TextFileKnowledgeSource(
file_paths=["docs/company.txt", "docs/products.txt"]
)
agent = Agent(
role="知识助手",
knowledge_sources=[knowledge]
)
PDF 知识源 #
python
from crewai.knowledge.source.pdf_knowledge_source import PDFKnowledgeSource
knowledge = PDFKnowledgeSource(
file_paths=["docs/manual.pdf"]
)
agent = Agent(
role="文档助手",
knowledge_sources=[knowledge]
)
Crew 级别知识源 #
python
crew = Crew(
agents=[agent1, agent2],
tasks=[task1, task2],
knowledge_sources=[knowledge1, knowledge2]
)
异步执行 #
异步 Crew 执行 #
python
import asyncio
from crewai import Crew
async def main():
crew = Crew(
agents=[agent],
tasks=[task]
)
result = await crew.akickoff(inputs={'topic': 'AI'})
print(result.raw)
asyncio.run(main())
异步任务执行 #
python
from crewai import Task
async_task = Task(
description="获取实时数据",
expected_output="数据报告",
agent=data_collector,
async_execution=True
)
sync_task = Task(
description="分析数据",
expected_output="分析报告",
agent=analyst,
context=[async_task]
)
批量异步执行 #
python
async def process_multiple():
inputs_list = [
{'topic': 'AI'},
{'topic': 'Blockchain'},
{'topic': 'IoT'}
]
results = await crew.akickoff_for_each(inputs=inputs_list)
for result in results:
print(result.raw)
asyncio.run(process_multiple())
流式输出 #
启用流式输出 #
python
crew = Crew(
agents=[agent],
tasks=[task],
stream=True
)
streaming = crew.kickoff(inputs={'topic': 'AI'})
for chunk in streaming:
print(chunk.content, end="", flush=True)
result = streaming.result
流式输出处理 #
python
def process_stream(crew, inputs):
streaming = crew.kickoff(inputs=inputs)
full_output = ""
for chunk in streaming:
full_output += chunk.content
yield chunk.content
return full_output
推理与规划 #
Agent 推理 #
python
from crewai import Agent
reasoning_agent = Agent(
role="战略规划师",
goal="分析复杂问题并制定计划",
backstory="经验丰富的战略顾问",
reasoning=True,
max_reasoning_attempts=3,
verbose=True
)
Crew 规划 #
python
from crewai import Crew
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, write_task],
planning=True,
planning_llm="gpt-4o",
verbose=True
)
规划流程 #
text
┌─────────────────────────────────────────────────────────────┐
│ 规划流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 分析阶段 │
│ └─> AgentPlanner 分析所有任务 │
│ └─> 识别任务依赖关系 │
│ └─> 确定执行顺序 │
│ │
│ 2. 计划生成 │
│ └─> 为每个任务生成详细计划 │
│ └─> 定义预期输出 │
│ └─> 分配资源 │
│ │
│ 3. 计划注入 │
│ └─> 计划添加到任务描述 │
│ └─> Agent 获得执行指导 │
│ │
│ 4. 执行阶段 │
│ └─> Agent 按计划执行 │
│ └─> 动态调整（如需要） │
│ │
└─────────────────────────────────────────────────────────────┘
多模态支持 #
启用多模态 #
python
from crewai import Agent
multimodal_agent = Agent(
role="视觉内容分析师",
goal="分析和处理文本及图像内容",
backstory="多模态 AI 专家",
multimodal=True,
verbose=True
)
处理图像 #
python
from crewai import Task
image_task = Task(
description="""
分析提供的图像：
1. 识别图像中的主要对象
2. 描述场景内容
3. 提取文本信息（如有）
""",
expected_output="图像分析报告",
agent=multimodal_agent
)
代码执行 #
安全模式 #
python
from crewai import Agent
safe_agent = Agent(
role="代码分析师",
goal="分析和执行代码",
backstory="安全的代码执行专家",
allow_code_execution=True,
code_execution_mode="safe",
max_execution_time=300
)
不安全模式 #
python
unsafe_agent = Agent(
role="代码执行器",
goal="直接执行代码",
backstory="受信任环境中的代码执行",
allow_code_execution=True,
code_execution_mode="unsafe"
)
代码执行任务 #
python
from crewai import Task
code_task = Task(
description="""
编写并执行 Python 代码：
1. 读取 CSV 文件
2. 数据清洗
3. 生成统计报告
""",
expected_output="执行结果和报告",
agent=safe_agent
)
上下文管理 #
自动上下文管理 #
python
agent = Agent(
role="分析师",
goal="分析大型文档",
backstory="文档分析专家",
respect_context_window=True
)
严格上下文限制 #
python
agent = Agent(
role="法律审查员",
goal="精确分析法律文档",
backstory="需要完整上下文的法律专家",
respect_context_window=False
)
使用 RAG 处理大数据 #
python
from crewai_tools import RagTool
rag_tool = RagTool()
agent = Agent(
role="知识库助手",
goal="从大型知识库检索信息",
tools=[rag_tool],
respect_context_window=True
)
回调系统 #
步骤回调 #
python
def step_callback(output):
print(f"[Step] Agent: {output.agent_role}")
print(f"[Step] Action: {output.action}")
print(f"[Step] Result: {output.text[:100]}...")
agent = Agent(
role="分析师",
step_callback=step_callback
)
任务回调 #
python
def task_callback(output):
print(f"[Task] Completed: {output.description}")
save_result(output.raw)
task = Task(
description="分析任务",
callback=task_callback
)
Crew 回调 #
python
crew = Crew(
agents=[agent],
tasks=[task],
step_callback=step_callback,
task_callback=task_callback
)
日期注入 #
启用日期注入 #
python
from crewai import Agent
agent = Agent(
role="市场分析师",
goal="跟踪市场动态",
backstory="时间敏感的分析专家",
inject_date=True,
date_format="%B %d, %Y"
)
日期格式 #
python
agent = Agent(
role="报告生成器",
inject_date=True,
date_format="%Y-%m-%d %H:%M:%S"
)
自定义模板 #
系统模板 #
python
agent = Agent(
role="助手",
system_template="""<|start_header_id|>system<|end_header_id|>
{{ .System }}<|eot_id|>"""
)
提示模板 #
python
agent = Agent(
role="助手",
prompt_template="""<|start_header_id|>user<|end_header_id|>
{{ .Prompt }}<|eot_id|>"""
)
响应模板 #
python
agent = Agent(
role="助手",
response_template="""<|start_header_id|>assistant<|end_header_id|>
{{ .Response }}<|eot_id|>"""
)
重放功能 #
查看任务输出 #
bash
crewai log-tasks-outputs
从特定任务重放 #
bash
crewai replay -t <task_id>
高级配置 #
速率限制 #
python
crew = Crew(
agents=[agent],
tasks=[task],
max_rpm=30
)
执行时间限制 #
python
agent = Agent(
role="分析师",
max_execution_time=300,
max_iter=25
)
重试配置 #
python
agent = Agent(
role="分析师",
max_retry_limit=3
)
下一步 #
现在你已经掌握了高级特性，接下来学习 最佳实践，了解生产级开发的技巧！
最后更新：2026-04-04