Day 12 / 共 60 天 · 阶段2 Agent 深入(收官)

LiteAgent:一个"不需要 Crew 也能跑"的轻量智能体

阶段2 前 5 天讲的都是"重型" Agent(Agent + CrewAgentExecutor + Task + Crew 那一整套)。但有时你只想"给个提示、带几个工具、要个结构化结果",不想组建 Crew。CrewAI 为此提供了轻量版 lite_agent.pyLiteAgent,输出用 lite_agent_output.pyLiteAgentOutput 统一封装。今天读它、和前 5 天的重型 Agent 逐点对比,并给出"何时用哪个"的选型建议,为阶段2 收官。

📍 你在 60 天里的位置(阶段2 Agent 深入 · 收官日)
阶段1 入门 D01-06 D07 全字段 D08 执行循环 D09 输出解析 D10 单步执行 D11 工具缓存 D12 LiteAgent 阶段3 Task
💡 先用一个类比兜住今天 重型 Agent 像一辆全配置的工程车:能进 Crew 车队、能被 Task 派活、能委派、能规划、能接记忆——功能全,但你得先"组队、派任务"才能开动。LiteAgent 像一辆共享单车:扫码即走,agent.kickoff("帮我查下天气") 一句就跑,不用组队、不用建 Task。它去掉了"车队协作"那套重装备,只保留"一个人 + 几个工具 + 一个循环"的核心。同一套 ReAct 内核,一个面向"多智能体协作",一个面向"单次直接调用"。
L01

痛点:不是所有活都值得组一个 Crew

🤔 痛点要让一个带工具的 Agent 干活,前 5 天的路径是:建 Agent → 建 Task(描述 + expected_output)→ 建 Crew(agents + tasks)→ crew.kickoff()。可如果我只想"给一句话、带一个搜索工具、要个 JSON 结果",这一整套组队仪式太重了——我没有"多个 Agent 协作"的需求,为什么要被迫走 Task/Crew 那套流程?有没有"一句话直接调用"的轻量入口?
💡 一句话本质 LiteAgent 是重型 Agent 的"精简直调版":保留 ReAct 核心(问 LLM → 工具 → 观察 → 循环,复用 Day 08/09 的同一批公共函数),但剥掉 Crew、Task、委派、Process 等"多智能体协作"的重装备,提供一个 kickoff(messages) 直接入口,返回一个统一的 LiteAgentOutput同一个内核,两种包装:一个为"协作",一个为"直调"。

类的定义与定位(lite_agent.py:187):

# lite_agent.py:187
class LiteAgent(FlowTrackable, BaseModel):
    """...
    Use ``Agent().kickoff(messages)`` instead, which provides the same ...
    """
大白话docstring 里提示:其实你平时用 Agent(...).kickoff("...") 就行——Agentkickoff 方法内部就会造一个 LiteAgent 来跑。也就是说 LiteAgent 常常是"重型 Agent 想轻量执行时的内部引擎"。理解它,你就理解了 CrewAI"一个 Agent 两种活法"的设计。
L02

字段对比:LiteAgent 减掉了什么

LiteAgent 的字段(lite_agent.py:212),和 Day 07 的重型 Agent 对照着看:

# lite_agent.py:212
id: UUID4 = Field(default_factory=uuid.uuid4, frozen=True)
role: str = Field(description="Role of the agent")            # 身份三件套还在
goal: str = Field(description="Goal of the agent")
backstory: str = Field(description="Backstory of the agent")
llm: str | BaseLLM | Any | None = Field(default=None, ...)   # 接大脑还在
tools: list[BaseTool] = Field(default_factory=list, ...)     # 工具还在
max_iterations: int = Field(default=15, ...)                 # 循环上限(注意默认 15,不是 25!)
max_execution_time: int | None = Field(default=None, ...)
respect_context_window: bool = Field(default=True, ...)      # 上下文压缩还在
use_stop_words: bool = Field(default=True, ...)
response_format: type[BaseModel] | None = Field(default=None, ...)  # 结构化输出
verbose: bool = Field(default=False, ...)
guardrail: Annotated[...] = Field(...)                       # 护栏还在
guardrail_max_retries: int = Field(default=3, ...)
memory: bool | Any | None = Field(default=None, ...)         # 可选记忆
能力重型 Agent(D07)LiteAgent
role/goal/backstory✅ 保留
llm / tools✅ 保留
循环上限max_iter=25max_iterations=15
guardrail 护栏✅ 保留
结构化输出需 Task 配response_format 直接给
Crew / Process❌ 去掉
allow_delegation 委派❌ 去掉
executor_class 可换执行器❌ 内建循环
planning 规划❌ 去掉
max_iterations=15边界注意:轻量版默认循环上限是 15,重型 Agent 是 25,字段名也不同(max_iterations vs max_iter)。看起来像小差异,实际用错会让你困惑"为什么它比预期早停"。
response_format 直接给轻量版把"要什么格式的结构化输出"直接做成字段,不用像重型那样在 Task 上配 output_pydantic。更适合"直调要 JSON"的场景。
没有 crew / process / delegation★核心区别:轻量版不参与多智能体协作。它就是"一个人干活",所以委派、流程、车队全没有。
💡 设计取舍①:为什么单独做一个 LiteAgent,而不是给 Agent 加个"轻量模式"开关? 朴素做法是给重型 Agent 加个 lite=True 参数,内部走简化分支。但源码另起一个类。原因:重型 Agent 的字段和方法深度耦合 Crew/Task/委派——它的很多逻辑假设"我在一个 Crew 里、有 Task 上下文"。硬塞一个轻量模式会让这个类到处是 if lite: ... else: ...,越来越难维护。把"直调"这个不同的使用形态抽成独立的、更小的类,各自内聚、互不拖累,代码反而更清晰。代价是有一点重复(两个类都有 role/goal/循环),但换来了各自的简单。这就是"组合优于开关参数"——用两个专注的类,而不是一个啥都能干的大类。
L03

kickoff:一句话直接开跑

入口 kickofflite_agent.py:477)——注意它直接吃字符串或消息列表:

# lite_agent.py:477
def kickoff(self, messages: str | list[LLMMessage],
            response_format: type[BaseModel] | None = None,
            input_files: dict[str, FileInput] | None = None) -> LiteAgentOutput:
    if self._memory is not None:              # 可选:把记忆能力包成工具挂上去
        ...
        self._parsed_tools = self._parsed_tools + parse_tools(memory_tools)
    agent_info = {"id": self.id, "role": self.role, "goal": self.goal, ...}
    try:
        self._iterations = 0                  # ① 循环计数归零
        self.tools_results = []
        self._messages = self._format_messages(messages, response_format=..., input_files=...)  # ② 组装消息
        self._inject_memory_context()         # ③ 注入记忆上下文(若有)
        return self._execute_core(agent_info=agent_info, response_format=response_format)  # ④ 进核心
    except Exception as e:
        ...
        crewai_event_bus.emit(self, event=LiteAgentExecutionErrorEvent(...))
        raise e
messages: str | list★入口直接吃一句话字符串("帮我查天气")或标准消息列表。不需要 Task 对象——这就是"轻"的体感。
① _iterations = 0和 Day 08 一样,每次执行重置圈数。
② _format_messages把用户的字符串/列表 + 系统提示(role/goal/backstory 拼的) + 可能的附件文件,组装成 LLM 的初始消息。
④ _execute_core真正的执行核心,含循环 + 结果封装 + 护栏(L05)。
emit 错误事件出错时通过事件总线发一个 LiteAgentExecutionErrorEvent(阶段4 D26 讲事件系统)再抛出——可观测性内建。
📝 真实值:最简用法 from crewai import Agent
a = Agent(role="天气助手", goal="回答天气", backstory="你熟悉各城市天气")
out = a.kickoff("北京今天多少度?") # 内部就走 LiteAgent
print(out.raw) # 一句话拿到结果,全程没建 Task、没建 Crew
L04

_invoke_loop:又是那个熟悉的 while

核心循环 _invoke_looplite_agent.py:860),和 Day 08 几乎是双胞胎:

# lite_agent.py:860
def _invoke_loop(self, response_model=None) -> AgentFinish:
    formatted_answer: AgentAction | AgentFinish | None = None
    while not isinstance(formatted_answer, AgentFinish):        # ★同样的循环条件
        try:
            if has_reached_max_iterations(self._iterations, self.max_iterations):
                formatted_answer = handle_max_iterations_exceeded(...)   # 同样的刹车
            enforce_rpm_limit(self.request_within_rpm_limit)            # 同样的限流
            answer = get_llm_response(llm=..., messages=self._messages, ...)  # 同样的问 LLM
            if isinstance(answer, BaseModel):
                formatted_answer = AgentFinish(thought="", output=answer, text=answer.model_dump_json())
                break
            formatted_answer = process_llm_response(cast(str, answer), self.use_stop_words)  # 同样的解析(D09)
            if isinstance(formatted_answer, AgentAction):
                tool_result = execute_tool_and_check_finality(   # 同样的执行工具
                    agent_action=formatted_answer, tools=self._parsed_tools,
                    agent_key=self.key, agent_role=self.role, agent=self.original_agent, crew=None)
                formatted_answer = handle_agent_action_core(   # 同样的把结果写回
                    formatted_answer=formatted_answer, tool_result=tool_result, show_logs=self._show_logs)
            self._append_message(formatted_answer.text, role="assistant")
        except OutputParserError as e:
            formatted_answer = handle_output_parser_exception(e=e, ...)   # 同样的自愈
    ...
while not ...AgentFinish★和 Day 08 一模一样的循环骨架。这就是"同一个 ReAct 内核"的直接证据。
has_reached_max_iterations / enforce_rpm_limit / get_llm_response / process_llm_response / execute_tool_and_check_finality / handle_output_parser_exception全是前几天见过的同一批公共函数!从 utilities/agent_utils 复用。轻量版没有重写循环,而是用同一批零件重新组装
crew=None执行工具时明确传 crew=None——因为 LiteAgent 不属于任何 Crew。这行小小的差异,正是"去协作化"的具体体现。
BaseModel 直接 finish如果 LLM 通过原生结构化输出直接返回了一个 Pydantic 对象,就直接包成 AgentFinish——支持 response_format 的结构化路径。
💡 "复用零件、重新组装"是本阶段最重要的一课对比 Day 08、Day 10、Day 12 三个循环:它们结构惊人地相似(都是 while not AgentFinish + 同一批 get_llm_response / process_llm_response / execute_tool_and_check_finality),差别只在"周边耦合什么"(大执行器耦合 Crew/Task,单步执行器隔离状态,LiteAgent 去掉协作)。把核心步骤抽成公共函数、让不同"执行形态"各自组装——这是 CrewAI 架构的精髓,也是你读任何成熟框架时该寻找的模式:先找"公共零件",再看"谁怎么组装"。
L05

结果封装与护栏:不合格就打回重做

_execute_corelite_agent.py:619)在循环之后做结果封装和护栏校验:

# lite_agent.py:619
def _execute_core(self, agent_info, response_format=None) -> LiteAgentOutput:
    ...
    agent_finish = self._invoke_loop(response_model=active_response_format)  # ① 跑循环
    ...
    # ② 尽力把输出转成 response_format 指定的 Pydantic 对象
    if isinstance(agent_finish.output, BaseModel):
        formatted_result = agent_finish.output
    elif active_response_format:
        try:
            formatted_result = active_response_format.model_validate_json(str(agent_finish.output))
        except ValidationError:
            ... # 退而用 Converter 让 LLM 再规整一次
    usage_metrics = self.llm.get_token_usage_summary() if isinstance(self.llm, BaseLLM) else ...
    output = LiteAgentOutput(raw=raw_output, pydantic=formatted_result,   # ③ 打包统一输出
                             agent_role=self.role, usage_metrics=..., messages=self._messages)
    if self._guardrail is not None:                                       # ④ 护栏校验
        guardrail_result = process_guardrail(output=output, guardrail=self._guardrail, ...)
        if not guardrail_result.success:
            if self._guardrail_retry_count >= self.guardrail_max_retries:
                raise Exception(f"Agent's guardrail failed validation after ... retries.")  # 重试用尽→抛
            self._guardrail_retry_count += 1
            self._messages.append({"role": "user", "content": guardrail_result.error or ...})  # 把失败原因喂回
            return self._execute_core(agent_info=agent_info)              # ★递归重跑一遍
    ...
    return output
② 尽力转结构化先试直接 model_validate_json;失败再用 Converter 让 LLM 把输出规整成目标格式。多级兜底,对应 Day 09 "尽力修"的同款哲学。
③ LiteAgentOutput把 raw 文本、pydantic 对象、token 用量、完整 messages 统一打包(L06 详解)。
④ 护栏不过 → 喂回原因 + 递归重跑★校验失败时,把失败原因作为一条 user 消息 append,然后 return self._execute_core(...) 递归重跑整个核心——相当于"这次没达标,把问题告诉它,让它带着反馈再做一遍"。
重试用尽 → raise递归有出口:_guardrail_retry_countguardrail_max_retries(默认 3)就抛异常,不会无限递归。
💡 设计取舍②:护栏为什么用"递归重跑 + 计数上限"而不是"内层 while 重试"? 护栏重试也可以写成一个 while retry < max: ... 的循环。源码却用递归调用 _execute_core。好处:递归天然复用了"跑循环 → 封装 → 校验"这一整套逻辑,每次重试都是完整、干净的一遍(把失败原因喂回后重新走全流程),代码不用为"重试"单独写一套。关键是递归必须有出口——这里靠 _guardrail_retry_count 计数 + guardrail_max_retries 上限保证不会无限递归。"递归 + 硬上限"是"带反馈重试"的一种优雅写法,但上限是命门——没有它就是定时炸弹(栈溢出/烧钱)。这也呼应 Day 08 的 max_iter任何"自动重试/循环"都必须配一个硬刹车。
L06

LiteAgentOutput:一个结果,多种取法

输出统一封装在 LiteAgentOutputlite_agent_output.py:30):

# lite_agent_output.py:30
class LiteAgentOutput(BaseModel):
    model_config = {"arbitrary_types_allowed": True}
    raw: str = Field(description="Raw output of the agent", default="")          # 原始文本
    pydantic: BaseModel | None = Field(description="Pydantic output ...", default=None)  # 结构化对象
    agent_role: str = Field(description="Role of the agent that produced this output")
    usage_metrics: dict[str, Any] | None = Field(default=None, ...)              # token 用量
    messages: list[LLMMessage] = Field(default_factory=list, ...)                # 完整对话
    plan: str | None = Field(default=None, ...)                                  # 若规划过:计划
    todos: list[TodoExecutionResult] = Field(default_factory=list, ...)          # 若规划过:各步结果
    replan_count: int = Field(default=0, ...)

# lite_agent_output.py:104
def __str__(self) -> str:
    return self.raw                            # print(output) 直接显示原始文本
raw永远有:模型输出的原始文本。__str__ 返回它,所以 print(output) 直接看到答案。
pydantic只有配了 response_format 且转换成功时才有:一个校验过的结构化对象。要类型安全的字段就取它。
usage_metrics这次跑了多少 token(成本可观测)。阶段8 讲 token/成本时会重逢。
messages完整对话历史。调试"它中间怎么想的、调了哪些工具"时非常有用。
plan / todos / replan_count若这次执行走了规划路径,这里记录计划文本和每步结果(TodoExecutionResultlite_agent_output.py:13)——把 Day 10 单步执行的产物汇总在此。
数据结构:LiteAgentOutput 一次结果的多个切面 LiteAgentOutput raw 原始文本(必有) pydantic 结构化(可选) usage_metrics token 成本 messages / todos 过程留痕 一次执行 → 一个对象,答案/结构/成本/过程各取所需
图注:不管你要"答案文本""结构化字段""成本"还是"调试过程",都从同一个输出对象取——统一封装的价值。
L07

选型建议 + 边界 + 阶段2 收官

🧭 选型:LiteAgent(直调)vs 重型 Agent+Crew(协作) 用 LiteAgent / agent.kickoff():单个 Agent、一次性问答、带几个工具、要结构化结果、不需要多 Agent 协作。像"函数调用"一样即调即用。
用重型 Agent + Task + Crew:多个 Agent 分工协作、任务之间有依赖/上下文传递、需要委派、需要 Process(顺序/层级)编排、需要 Crew 级记忆。
一句话:"一个人的活"用轻量直调,"一个团队的活"用重型编排。
⚠️ 边界:messages 是 property,别当普通属性乱塞 LiteAgent 对外的 messageslite_agent.py:468)和 iterations:473)是只读 property,内部真正的状态存在 self._messages / self._iterations(带下划线的私有属性)。为什么这么设计?因为 Pydantic 模型的公开字段会被校验/序列化,而"执行中的临时对话状态"不该混进模型的持久字段里。_下划线私有属性存运行时状态 + property 只读暴露,既让外部能看(调试),又防止外部乱改破坏循环。如果你想往里塞消息,要走它的方法(如 _append_message),别直接赋值——这类"看着是属性其实是 property"的地方,是读 Pydantic 模型时容易踩的坑。

👶 小白:既然 LiteAgent 这么方便,为什么不全用它?

👨‍🏫 老师:因为它放弃了协作能力。CrewAI 的核心价值恰恰是"多个角色分工协作"(调研员 + 撰稿人 + 校对员流水线)——这些必须靠重型 Agent + Task + Crew + Process 才能编排。LiteAgent 是给"单点、简单、直接"场景的快捷方式,不是要取代重型体系。它俩是互补的两个入口,不是新旧替代。看懂这一点,你就理解了整个阶段2:Agent 既能"独当一面"(LiteAgent),也能"团队作战"(重型 + Crew),共享同一个 ReAct 内核。

🧠 阶段2 收官自测(D07–D12 串起来)

  • D07:Agent 是配置对象,字段分哪五组?为什么配置与执行分离?
  • D08:执行器的本质是 while not AgentFinish;原生 vs 文本怎么分流?
  • D09:parser 把文本翻译成 AgentAction/AgentFinish;解析失败如何自愈?
  • D10:StepExecutor 单步隔离执行;失败为何返回 StepResult 而非抛异常?
  • D11:工具缓存 key=工具名+输入;读写锁为何优于普通锁?
  • D12:LiteAgent 与重型 Agent 共享内核、去掉协作;何时用哪个?
  • 贯穿主线:三个循环(D08/D10/D12)复用同一批公共函数——"抽零件、再组装"。

✋ 10 分钟动手

P=lib/crewai/src/crewai
sed -n '187,289p' $P/lite_agent.py           # LiteAgent 字段
sed -n '477,546p' $P/lite_agent.py           # kickoff
sed -n '860,931p' $P/lite_agent.py           # _invoke_loop(对照 Day 08)
cat -n            $P/lite_agent_output.py     # LiteAgentOutput
# 亲手跑一次轻量直调
python -c "
from crewai import Agent
a = Agent(role='助手', goal='简洁回答', backstory='你答得又快又准')
out = a.kickoff('用一句话解释什么是 ReAct')
print('raw =', out.raw)
print('role =', out.agent_role)
"
下阶段预告 · 阶段3 Task 深入(D13–D18):Agent 讲透了,接下来看它执行的"活"——Task。Day 13 起逐字段拆 task.py:任务描述、expected_outputTaskOutput 与护栏、结构化输出、context 依赖、异步与条件任务。你会看到今天的 guardrailresponse_format 在 Task 层如何再次出现并组合。
← Day 11 工具缓存 Day 13 · Task 模型全字段 →