垦荒学园
  • 中文

      • Crew 与 Process

      Crew 是一组 Agent 加上一组 Task，以及一条 Process（执行策略）。调用 crew.kickoff(inputs=...) 启动；异步用 kickoff_async()。

      生产应用里，Crew 通常出现在 Flow 的某一个 @listen 步骤 中，而不是作为整个程序的唯一入口。见 简介 与 Flows。

      创建与运行

      from crewai import Agent, Task, Crew, Process

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True,
    memory=True,       # 统一记忆，见 Memory 章
    planning=False,    # True 时先生成任务计划再执行
)

output = crew.kickoff(inputs={"topic": "CrewAI Flows"})
print(output.raw)
print(output.tasks_output)
print(output.token_usage)

      CrewOutput：raw、pydantic、json_dict、tasks_output、token_usage。需要 JSON 时在最后那个 Task 上设 output_pydantic / output_json。

      其它常用参数：cache（工具结果缓存，默认 True）、max_rpm、embedder（记忆用，默认 OpenAI）、manager_llm / manager_agent、output_log_file、stream、checkpoint。

      Process.sequential

      Process 是枚举，合法值是 sequential 与 hierarchical。

      顺序流程按 tasks 列表执行，上一任务输出作为后续上下文。这是默认值，也是大多数教程的起点。

      crew = Crew(
    agents=my_agents,
    tasks=my_tasks,
    process=Process.sequential,
)

      用 Task 的 context=[other_task] 精确指定依赖，而不是假设「列表顺序 = 唯一数据流」。

      Process.hierarchical

      层级流程模仿「经理分配工作」：任务不必预先绑死 Agent。必须提供 manager_llm 或 manager_agent。经理负责规划、委派、验收。

      crew = Crew(
    agents=my_agents,
    tasks=my_tasks,
    process=Process.hierarchical,
    manager_llm="openai/gpt-4o",
    # 或 manager_agent=my_manager_agent
)

      JSONC："process": "hierarchical"，并设置 manager_llm 或 manager_agent（可指向未列入顶层 agents 数组的 jsonc 文件）。

      层级流程 token 更多、更难调试。先把顺序流程跑稳，再在「任务动态分派」场景启用。

      JSONC 与经典脚手架

      默认 crewai create crew 生成 JSON-first：crew.jsonc + agents/*.jsonc，crewai run 直接加载。

      {
  "name": "Market Research Crew",
  "agents": ["researcher", "analyst"],
  "process": "sequential",
  "verbose": true,
  "memory": true,
  "tasks": [],
  "inputs": { "topic": "AI Agents" }
}

      crewai create crew <name> --classic 使用 crew.py 与 YAML，以及文档中的 @CrewBase、@agent、@task、@crew（仅经典路径；不要把这些装饰器用在 Flow 上）。JSON 里的 "custom:" 工具和 {"python": "module.attribute"} 会执行本地 Python，只运行你信任的项目。

      加载 JSON Crew：

      from pathlib import Path
from crewai.project import load_crew

crew, default_inputs = load_crew(Path("crew.jsonc"))
result = crew.kickoff(inputs={**default_inputs, "topic": "AI Agents"})

      检查点与日志

      checkpoint=True（或 CheckpointConfig）可在任务完成后保存状态，中断后从断点继续。output_log_file=True 写入 logs.txt；路径以 .json 结尾则为 JSON 日志。

      下一步

      评论