看 TaskOutput 的字段声明（tasks/task_output.py:14）：

# tasks/task_output.py:14
class TaskOutput(BaseModel):
    """Class that represents the result of a task."""
    description: str = Field(description="Description of the task")
    name: str | None = Field(default=None)
    expected_output: str | None = Field(default=None)
    summary: str | None = Field(default=None)          # 自动从 description 截取
    raw: str = Field(default="")                        # ★永远有：原始文本
    pydantic: BaseModel | None = Field(default=None)    # 配了 output_pydantic 才有
    json_dict: dict[str, Any] | None = Field(default=None)  # 配了 output_json 才有
    agent: str = Field(description="Agent that executed the task")
    output_format: OutputFormat = Field(default=OutputFormat.RAW)  # 主推哪副面孔
    messages: list[LLMMessage] = Field(default_factory=list)

TaskOutput 还有三个贴心方法。先看 summary——它在对象造好后自动从描述截前 10 个词（tasks/task_output.py:50）：

# tasks/task_output.py:50
@model_validator(mode="after")
def set_summary(self) -> TaskOutput:
    excerpt = " ".join(self.description.split(" ")[:10])
    self.summary = f"{excerpt}..."
    return self

再看 json 属性和 __str__（tasks/task_output.py:61、tasks/task_output.py:99）：

# tasks/task_output.py:61
@property
def json(self) -> str | None:
    if self.output_format != OutputFormat.JSON:
        raise ValueError("Invalid output format requested. ...set the output_json...")
    return json.dumps(self.json_dict)

# tasks/task_output.py:99
def __str__(self) -> str:
    if self.pydantic:   return str(self.pydantic)     # 优先对象
    if self.json_dict:  return str(self.json_dict)    # 再 json
    return self.raw                                   # 最后兜底文本

护栏在 Task 上有两个字段：单个 guardrail 和一串 guardrails（task.py:246、task.py:257），还有重试次数 guardrail_max_retries（task.py:273，默认 3）。它可以是一个函数，也可以是一句话描述：

# 用法示例（两种写法）
# ① 写个函数：返回 (是否通过, 结果或错误说明)
def no_english(output: TaskOutput) -> tuple[bool, Any]:
    if any(c.isalpha() and c.isascii() for c in output.raw):
        return (False, "输出里不能出现英文字母")
    return (True, output.raw)
Task(..., guardrail=no_english)

# ② 写一句话，让 LLM 当裁判（见 L05）
Task(..., guardrail="输出必须是纯中文，且不超过 100 字", agent=writer)

当你把 guardrail 写成字符串时，Task 的 after 校验器会把它包装成一个 LLMGuardrail（task.py:377）：

# task.py:377
@model_validator(mode="after")
def ensure_guardrail_is_callable(self) -> Task:
    if callable(self.guardrail):
        self._guardrail = self.guardrail            # 本来就是函数：直接用
    elif isinstance(self.guardrail, str):           # 是字符串：包成 LLM 裁判
        from crewai.tasks.llm_guardrail import LLMGuardrail
        if self.agent is None:
            raise ValueError("Agent is required to use LLMGuardrail")
        if not isinstance(self.agent.llm, BaseLLM):
            raise ValueError("Agent must have a BaseLLM instance to use LLMGuardrail")
        self._guardrail = cast(
            GuardrailCallable,
            LLMGuardrail(description=self.guardrail, llm=self.agent.llm),
        )
    return self

而 LLMGuardrail 本身，就是"拿另一个 Agent 当裁判"：把你的产出和规则一起丢给 LLM，让它判合不合格（tasks/llm_guardrail.py:69、tasks/llm_guardrail.py:98）：

# tasks/llm_guardrail.py:69
def _validate_output(self, task_output: TaskOutput) -> LiteAgentOutput:
    agent = Agent(role="Guardrail Agent", goal="Validate the output of the task",
                  backstory="You are a expert at validating the output of a task...",
                  llm=self.llm)
    query = f"""
    Ensure the following task result complies with the given guardrail.
    Task result: {task_output.raw}
    Guardrail: {self.description}
    - Confirm if the Task result complies with the guardrail.
    - If not, provide clear feedback explaining what is wrong ...
    """
    kickoff_result = agent.kickoff(query, response_format=LLMGuardrailResult)  # ★结构化返回
    ...

# tasks/llm_guardrail.py:98
def __call__(self, task_output: TaskOutput) -> tuple[bool, Any]:
    result = self._validate_output(task_output)
    if result.pydantic.valid:
        return True, task_output.raw           # 合格：放行原文
    return False, result.pydantic.feedback     # 不合格：返回 LLM 给的反馈

护栏真正的执行在 _invoke_guardrail_function（task.py:1246）。这是今天的心脏——一个"校验→不过就带反馈重跑→再校验"的循环：

# task.py:1246（骨架）
def _invoke_guardrail_function(self, task_output, agent, tools, guardrail, guardrail_index=None):
    current_retry_count = ...                       # 该护栏已重试几次
    max_attempts = self.guardrail_max_retries + 1   # 允许尝试的总次数

    for attempt in range(max_attempts):
        guardrail_result = process_guardrail(       # ★跑一次质检（L07）
            output=task_output, guardrail=guardrail, retry_count=current_retry_count,
            event_source=self, from_task=self, from_agent=agent)

        if guardrail_result.success:                # 通过！
            if guardrail_result.result is None:
                raise Exception("Task guardrail returned None as result. ...")
            if isinstance(guardrail_result.result, str):
                task_output.raw = guardrail_result.result           # 护栏可顺手改文本
                task_output.pydantic, task_output.json_dict = self._export_output(...)
            elif isinstance(guardrail_result.result, TaskOutput):
                task_output = guardrail_result.result               # 或直接换整个产出
            return task_output

        if attempt >= self.guardrail_max_retries:   # 重试用尽 → 放弃报错
            raise Exception(f"Task failed guardrail validation after "
                            f"{self.guardrail_max_retries} retries. Last error: ...")

        self.retry_count += 1
        context = I18N_DEFAULT.errors("validation_error").format(  # ★把错误塞进上下文
            guardrail_result_error=guardrail_result.error, task_output=task_output.raw)
        result = agent.execute_task(task=self, context=context, tools=tools)  # 带反馈重做
        task_output = TaskOutput(...raw=result...)  # 重新包装，进下一轮 for 再校验

所有护栏——无论函数还是 LLM——都必须返回 (bool, Any) 元组。process_guardrail 把这个元组转成结构化的 GuardrailResult（utilities/guardrail.py:123、utilities/guardrail.py:105）：

# utilities/guardrail.py:123
def process_guardrail(output, guardrail, retry_count, ...) -> GuardrailResult:
    if not isinstance(output, (TaskOutput, LiteAgentOutput)):
        raise TypeError("Output must be a TaskOutput or LiteAgentOutput")
    ...
    crewai_event_bus.emit(event_source, started_event)   # 发"护栏开始"事件
    result = guardrail(output)                            # ★调护栏，拿 (bool, Any)
    guardrail_result = GuardrailResult.from_tuple(result) # 转结构化
    crewai_event_bus.emit(event_source, LLMGuardrailCompletedEvent(...))  # 发"护栏完成"事件
    return guardrail_result

# utilities/guardrail.py:105
@classmethod
def from_tuple(cls, result: tuple[bool, Any | str]) -> Self:
    success, data = result
    return cls(success=success,
               result=data if success else None,     # 成功：data 是结果
               error=data if not success else None)   # 失败：data 是错误说明

✋ 10 分钟动手

P=lib/crewai/src/crewai
# 1. TaskOutput 三副面孔 + 便利方法
sed -n '14,105p' $P/tasks/task_output.py

# 2. 字符串护栏 → LLMGuardrail
sed -n '377,396p' $P/task.py
sed -n '49,120p' $P/tasks/llm_guardrail.py

# 3. 带反馈重试循环
sed -n '1246,1353p' $P/task.py

# 4. (bool,Any) 契约 → GuardrailResult
sed -n '105,187p' $P/utilities/guardrail.py