LiteAgent:一个"不需要 Crew 也能跑"的轻量智能体
阶段2 前 5 天讲的都是"重型" Agent(Agent + CrewAgentExecutor + Task + Crew 那一整套)。但有时你只想"给个提示、带几个工具、要个结构化结果",不想组建 Crew。CrewAI 为此提供了轻量版 lite_agent.py 的 LiteAgent,输出用 lite_agent_output.py 的 LiteAgentOutput 统一封装。今天读它、和前 5 天的重型 Agent 逐点对比,并给出"何时用哪个"的选型建议,为阶段2 收官。
Agent 像一辆全配置的工程车:能进 Crew 车队、能被 Task 派活、能委派、能规划、能接记忆——功能全,但你得先"组队、派任务"才能开动。LiteAgent 像一辆共享单车:扫码即走,agent.kickoff("帮我查下天气") 一句就跑,不用组队、不用建 Task。它去掉了"车队协作"那套重装备,只保留"一个人 + 几个工具 + 一个循环"的核心。同一套 ReAct 内核,一个面向"多智能体协作",一个面向"单次直接调用"。痛点:不是所有活都值得组一个 Crew
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 ...
"""
Agent(...).kickoff("...") 就行——Agent 的 kickoff 方法内部就会造一个 LiteAgent 来跑。也就是说 LiteAgent 常常是"重型 Agent 想轻量执行时的内部引擎"。理解它,你就理解了 CrewAI"一个 Agent 两种活法"的设计。字段对比: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=25 | max_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★核心区别:轻量版不参与多智能体协作。它就是"一个人干活",所以委派、流程、车队全没有。Agent 加个 lite=True 参数,内部走简化分支。但源码另起一个类。原因:重型 Agent 的字段和方法深度耦合 Crew/Task/委派——它的很多逻辑假设"我在一个 Crew 里、有 Task 上下文"。硬塞一个轻量模式会让这个类到处是 if lite: ... else: ...,越来越难维护。把"直调"这个不同的使用形态抽成独立的、更小的类,各自内聚、互不拖累,代码反而更清晰。代价是有一点重复(两个类都有 role/goal/循环),但换来了各自的简单。这就是"组合优于开关参数"——用两个专注的类,而不是一个啥都能干的大类。kickoff:一句话直接开跑
入口 kickoff(lite_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 Agenta = Agent(role="天气助手", goal="回答天气", backstory="你熟悉各城市天气")out = a.kickoff("北京今天多少度?") # 内部就走 LiteAgentprint(out.raw) # 一句话拿到结果,全程没建 Task、没建 Crew_invoke_loop:又是那个熟悉的 while
核心循环 _invoke_loop(lite_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 的结构化路径。while not AgentFinish + 同一批 get_llm_response / process_llm_response / execute_tool_and_check_finality),差别只在"周边耦合什么"(大执行器耦合 Crew/Task,单步执行器隔离状态,LiteAgent 去掉协作)。把核心步骤抽成公共函数、让不同"执行形态"各自组装——这是 CrewAI 架构的精髓,也是你读任何成熟框架时该寻找的模式:先找"公共零件",再看"谁怎么组装"。结果封装与护栏:不合格就打回重做
_execute_core(lite_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_count 到 guardrail_max_retries(默认 3)就抛异常,不会无限递归。while retry < max: ... 的循环。源码却用递归调用 _execute_core。好处:递归天然复用了"跑循环 → 封装 → 校验"这一整套逻辑,每次重试都是完整、干净的一遍(把失败原因喂回后重新走全流程),代码不用为"重试"单独写一套。关键是递归必须有出口——这里靠 _guardrail_retry_count 计数 + guardrail_max_retries 上限保证不会无限递归。"递归 + 硬上限"是"带反馈重试"的一种优雅写法,但上限是命门——没有它就是定时炸弹(栈溢出/烧钱)。这也呼应 Day 08 的 max_iter:任何"自动重试/循环"都必须配一个硬刹车。LiteAgentOutput:一个结果,多种取法
输出统一封装在 LiteAgentOutput(lite_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若这次执行走了规划路径,这里记录计划文本和每步结果(TodoExecutionResult,lite_agent_output.py:13)——把 Day 10 单步执行的产物汇总在此。选型建议 + 边界 + 阶段2 收官
agent.kickoff():单个 Agent、一次性问答、带几个工具、要结构化结果、不需要多 Agent 协作。像"函数调用"一样即调即用。用重型 Agent + Task + Crew:多个 Agent 分工协作、任务之间有依赖/上下文传递、需要委派、需要 Process(顺序/层级)编排、需要 Crew 级记忆。
一句话:"一个人的活"用轻量直调,"一个团队的活"用重型编排。
messages(lite_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)
"
Task。Day 13 起逐字段拆 task.py:任务描述、expected_output、TaskOutput 与护栏、结构化输出、context 依赖、异步与条件任务。你会看到今天的 guardrail、response_format 在 Task 层如何再次出现并组合。