Chapter 5 · Crew Assembly and Execution Flow

Chapter goals: Master the Crew object assembly and execution API (kickoff / kickoff_for_each / async variants), read the full CrewOutput structure, and learn to use inputs template variables for batch tasks and replay to re-run from a specified task.

5.1 Assembling a Proper Crew

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

deepseek = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.7,
)

researcher = Agent(role="Researcher", goal="Produce structured research points",
                   backstory="Ten years of industry research experience.", llm=deepseek)
writer = Agent(role="Writer", goal="Expand points into a fluent article",
               backstory="Senior tech column author.", llm=deepseek)

t1 = Task(description="Research '{topic}' and summarize 5 key points.",
          expected_output="Numbered list, each item under 30 characters.", agent=researcher)
t2 = Task(description="Write a 300-word essay based on the research.",
          expected_output="Markdown essay with a title and three subsections.", agent=writer)

crew = Crew(
    agents=[researcher, writer],
    tasks=[t1, t2],
    process=Process.sequential,   # Sequential execution (default)
    verbose=True,
)

Assembly rules:

• agents and tasks are two lists; each task binds its executor via the agent= field — unbound tasks will error at startup;
• process=Process.sequential executes tasks in definition order; hierarchical mode Process.hierarchical is covered in Chapter 9;
• output_log_file="crew_log.json" writes the full execution log to disk (supports .txt and .json) — essential for production debugging.

5.2 kickoff: Single Execution and Template Interpolation

result = crew.kickoff(inputs={"topic": "Large Model Inference Optimization"})
print(result.raw)          # Output text of the final task (t2)

The inputs dict replaces {topic} placeholders in all task texts. The same crew definition, a different inputs — run different topics — "separation of definition and data" is the first step of Crew engineering.

When there are no placeholders, pass kickoff() or kickoff(inputs={}).

5.3 kickoff_for_each: Batch Execution

Want to generate an essay for each of 10 topics? No need to loop calling kickoff repeatedly — the official API provides a batch interface:

topics = [
    {"topic": "Large Model Inference Optimization"},
    {"topic": "Multi-Agent Collaboration"},
    {"topic": "RAG Retrieval Augmentation"},
]

# Sequential batch: run the full crew for each input
results = crew.kickoff_for_each(inputs=topics)

for r in results:
    print("----")
    print(r.raw[:100])

For high-concurrency scenarios, use the native async version (officially recommended for high-throughput workloads):

import asyncio

async def main():
    results = await crew.akickoff_for_each(inputs=topics)   # Native async batch
    for r in results:
        print(r.raw[:80])

asyncio.run(main())

Similarly, single execution also has an async variant: await crew.akickoff(inputs={...}).

5.4 Reading CrewOutput

kickoff returns a structured CrewOutput object:

result = crew.kickoff(inputs={"topic": "AI Safety"})

print(result.raw)            # Final output原文 (default format)
print(result.tasks_output)   # List of TaskOutput for each task
print(result.token_usage)    # Token usage statistics

# Review intermediate products task by task — a debugging powerhouse
for i, t_out in enumerate(result.tasks_output, 1):
    print(f"Task {i}: {t_out.raw[:60]}...")

# The crew object also has cumulative usage metrics
print(crew.usage_metrics)

Attribute	Description
raw	Final output string (last task's output)
pydantic	Structured Pydantic object (requires output_pydantic config, see Chapter 12)
json_dict	JSON dict output (requires output_json config)
tasks_output	List[TaskOutput], each task's raw/structured output
token_usage	Token stats for this run

str(result) priority order: pydantic > json > raw.

5.5 replay: Re-run from a Specified Task

The most painful scenario when debugging long flows: task 5 failed, the results from the first 4 tasks are clearly still there — do you really have to re-run everything? CrewAI's replay mechanism saves the task outputs from the most recent kickoff locally, allowing you to resume from a specified task:

# CLI way: re-run from a specified task_id (task_id can be obtained from logs)
crewai replay -t <task_id>

You can also call crew.replay(task_id=...) in code. Applicable scenarios:

• Upstream tasks produced good output, only want to retry the failed task;
• Adjusted the prompt of a middle task and want to verify its impact on downstream without re-running earlier tasks.

TIP

replay depends on the locally saved most recent run record, so it only works for "the crew just run." For cross-session breakpoint recovery, use the checkpoint mechanism (covered in the deployment section of Chapter 22).

5.6 Chapter Summary

• Crew = agents + tasks + process; each task must bind an agent; sequential executes in definition order;
• kickoff(inputs={...}) uses template interpolation to achieve "one definition, multiple datasets";
• Batch with kickoff_for_each, high concurrency with native async akickoff / akickoff_for_each;
• CrewOutput.raw / tasks_output / token_usage are the result interpretation trio; crew.usage_metrics gives the cumulative bill;
• crewai replay -t <task_id> resumes from a specified task — no need to start over for long flow debugging.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What is result.raw returned by crew.kickoff()?

2. To run the same crew for 50 customers, which API is most appropriate?

3. Task 5 failed but the first 4 results are good; you want to resume from task 5. What should you use?

4. What is the correct way to check how many tokens this run consumed?

🛠️ Hands-On Exercises

1. Refactor the Chapter 2 crew into a batch version: prepare an input list of 3 different topics, run with kickoff_for_each, and print the first 80 characters of each essay.
2. Write code to iterate result.tasks_output, saving each task's output separately as output/task_1.md and output/task_2.md.
3. Intentionally change the last task's expected_output to mutually contradictory requirements to cause failure, then use replay to re-run from that task and fix it; record how to obtain the task_id.

A single crew is now mastered. Next, deep dive into LLM configuration and master all third-party model integration parameters.