Chapter 12 · Output Processing: JSON, Pydantic, and Data Passing

Chapter goals: Master structured output with output_pydantic / output_json, learn to retrieve each task's product via CrewOutput.tasks_output, understand implicit and explicit data flow between tasks, use usage_metrics for cost accounting, and build output post-processing pipelines.

12.1 Why Structured Output Is Needed

crew.kickoff() returns raw by default — a free-form text string. Comfortable for humans, unreliable for programs. LLM output Markdown heading hierarchy and list bullets vary every time; direct split() parsing is brittle. Structured output makes the crew's product directly into a type-checked Python object.

12.2 output_pydantic and output_json

import os

from crewai import Agent, Crew, Process, Task, LLM
from pydantic import BaseModel, Field

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.2,     # low temperature recommended for structured extraction
)

analyst = Agent(
    role="Financial Analyst",
    goal="Precisely extract key metrics from financial report text",
    backstory="A rigorous financial data expert who outputs only facts.",
    llm=llm,
)


# Define the expected output structure with Pydantic
class FinancialSummary(BaseModel):
    company: str = Field(..., description="Company name")
    revenue_yi: float = Field(..., description="Revenue in hundred millions of CNY")
    yoy_growth: float = Field(..., description="Year-over-year revenue growth rate (decimal, e.g. 0.15)")
    highlights: list[str] = Field(..., description="Up to 3 operational highlights")


task = Task(
    description="Read the following financial report summary and extract key metrics:\n{report_text}",
    expected_output="Structured data conforming to the schema",
    agent=analyst,
    output_pydantic=FinancialSummary,   # the key parameter
)

crew = Crew(agents=[analyst], tasks=[task], process=Process.sequential)
result = crew.kickoff(inputs={"report_text": "Company X reported 2025 revenue of 12 billion CNY, up 18% YoY..."})

# result.pydantic is a validated model instance, usable directly as a Python object
summary: FinancialSummary = result.pydantic
print(summary.company)          # attribute access, no string parsing needed
print(f"Growth rate: {summary.yoy_growth:.1%}")

# Or get dict / JSON string forms
print(result.json_dict)         # dict form (only populated when structured output is configured)

output_json works similarly (pass a Pydantic model or dict structure), with the difference that the output is exposed as a JSON dict in json_dict. Prefer output_pydantic when type checking and method reuse are needed.

populated only when configured

Official docs are explicit: TaskOutput.raw is always populated, but pydantic and json_dict are non-empty only when the corresponding parameter was configured on the task. Downstream code must defend against None values.

12.3 tasks_output: Retrieving Each Task's Product

CrewOutput holds more than the final result — tasks_output preserves each task's individual product:

result = crew.kickoff()

# Final output = last task's output
print("Final result:", result.raw)

# Inspect intermediate tasks one by one — key debugging technique for pipelines
for i, out in enumerate(result.tasks_output, 1):
    print(f"[Task {i}] agent={out.agent}")
    print(f"  raw first 60 chars: {out.raw[:60]}...")
    if out.pydantic:                       # if this task was configured with output_pydantic
        print(f"  structured fields: {list(out.pydantic.model_fields)}")

# Token usage stats (see 12.5 for details)
print(crew.usage_metrics)

The most common pattern is iterating tasks_output as shown above. Intermediate tasks can each independently configure output_pydantic, achieving a pipeline where "every step is strongly typed."

12.4 Data Passing Between Tasks: Implicit Flow vs Explicit Declaration

Under sequential flow, the previous task's output automatically becomes the next task's context (covered in Chapter 5). This is implicit passing:

t1 = Task(description="Research competitor pricing strategies", expected_output="pricing key points list", agent=researcher)
t2 = Task(
    description="Based on the research results, provide our pricing recommendations",   # "research results" auto-visible
    expected_output="pricing recommendation table",
    agent=strategist,
)

Implicit passing is convenient, but has two pitfalls: context grows longer = more tokens; and t2 sees t1's full raw output, not a distilled version. Explicit passing uses context to precisely control information flow:

# context explicitly declares dependencies: only the specified task outputs enter this task's context
report_task = Task(
    description="Synthesize the analysis and pricing conclusions into a management report",
    expected_output="one-page decision report",
    agent=writer,
    context=[analysis_task, pricing_task],   # feed only these two, not the raw research全文
)

One step further: combine into a "structured relay" pattern — upstream produces a Pydantic object, downstream task description references its fields:

class PricingAdvice(BaseModel):
    suggested_price: float
    rationale: str

pricing_task = Task(
    description="Provide product pricing recommendations",
    expected_output="structured pricing advice",
    agent=strategist,
    output_pydantic=PricingAdvice,
)

# Interpolate upstream structured conclusion in downstream description (rendered by framework after kickoff)
launch_task = Task(
    description="Draft launch copy based on pricing advice: {advice}",
    expected_output="launch copy under 100 words",
    agent=copywriter,
)
crew.kickoff(inputs={"advice": f"{pricing_task.output.raw}"})

12.5 usage_metrics: Cost Accounting for AI Pipelines

crew.kickoff()
metrics = crew.usage_metrics
print(metrics)
# Includes successful request count and token stats:
# total_tokens is the billing口径 = prompt_tokens + completion_tokens
# (cached_prompt_tokens and other sub-fields are already included in total, not added separately)

In production, pipe usage_metrics into your monitoring system (paired with Chapter 21's observability setup), aggregate by day/team, and catch cost overruns before they spiral. Hierarchical flow cost assessment (Chapter 9) especially depends on this number.

12.6 Output Post-Processing Pipeline

Recommended project layering: crew handles "generation," plain Python code handles "refinement and persistence," keeping the two decoupled:

from datetime import datetime
import json
from pathlib import Path


def post_process(summary: FinancialSummary) -> dict:
    """Business validation and enrichment on structured output; raise on failure, never silently pass."""
    if summary.yoy_growth < -0.5:
        raise ValueError(f"{summary.company} growth rate is异常: {summary.yoy_growth:.0%}. Requires manual review.")
    return {
        **summary.model_dump(),
        "captured_at": datetime.now().isoformat(),
        "reviewed": False,
    }


result = crew.kickoff(inputs={"report_text": ...})
record = post_process(result.pydantic)      # validate + enrich
Path("out").mkdir(exist_ok=True)
Path(f"out/{record['company']}.json").write_text(
    json.dumps(record, ensure_ascii=False, indent=2)
)

Key points: Pydantic validation catches "wrong format," business validation (e.g., growth rate thresholds) catches "right format but absurd values." Both gates must pass before persisting or handing off to downstream systems.

Chapter Summary

• Task(output_pydantic=Model) turns output into a validated Python object; output_json provides a dict form;
• pydantic / json_dict are populated only when the task configured the corresponding parameter — downstream code must defend against None;
• CrewOutput.tasks_output retrieves each task's product individually — essential for multi-step pipeline debugging;
• Sequential flow passes full context by default; context=[...] explicitly narrows the information flow, saving tokens and reducing noise;
• usage_metrics provides billing口径 token stats (total = prompt + completion);
• Post-processing belongs outside the crew: format validation + business validation double gate before persisting.

🧪 随堂测验

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

1. After configuring output_pydantic=FinancialSummary on a task, how do you get the validated object?

2. If a task has no structured output parameter configured, what is the value of its TaskOutput.json_dict?

3. Regarding inter-task data passing, what does context=[a_task, b_task] do?

4. Regarding usage_metrics, which statement is correct?

🛠️ Hands-On Exercises

1. Give each of three tasks in a pipeline a different output_pydantic model; run it and iterate tasks_output printing each step's structured fields.
2. Convert a 5-task sequential crew to use context to narrow the information flow; compare usage_metrics token differences before and after the change.
3. Implement the post-processing pipeline from 12.6: design at least two business validation rules (e.g., value thresholds, required fields) for your scenario, and intentionally construct a violating input to verify the gate catches it.

The single-crew "internal skills" are now complete. Next enters quality assurance: Chapter 13 · Planning, Iteration, and Error Handling.