一个标准项目的样子（crewai create crew 生成）：

@CrewBase
class MyCrew:
    agents_config = "config/agents.yaml"   # 花名册
    tasks_config  = "config/tasks.yaml"    # 任务单

    @agent
    def researcher(self) -> Agent:
        return Agent(config=self.agents_config["researcher"])  # 只引用 YAML

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config["research_task"])

    @crew
    def crew(self) -> Crew:
        return Crew(agents=self.agents, tasks=self.tasks, process=Process.sequential)

装饰器本体极简（annotations.py:66）：

# project/annotations.py:66
def task(meth: Callable[P, TaskResultT]) -> TaskMethod[P, TaskResultT]:
    """Marks a method as a crew task."""
    return TaskMethod(memoize(meth))          # 包一层"任务标记类" + memoize

# :78
def agent(meth: Callable[P, R]) -> AgentMethod[P, R]:
    """Marks a method as a crew agent."""
    return AgentMethod(memoize(meth))         # 包一层"Agent 标记类" + memoize

# :90 def llm(...)  :126 def tool(...)  :138 def callback(...)  同理

标记类是啥？看 wrappers.py 里同类结构（此处以 hook 方法为例，agent/task 标记类形态一致）——核心就是一个带 is_xxx=True 属性、能当方法用的包装对象：

# 每个标记类都携带一个布尔属性做"身份牌"，例如：
#   AgentMethod.is_agent = True
#   TaskMethod.is_task   = True
# 后面元类就是靠 hasattr(method, "is_agent") 把它们筛出来

@CrewBase 不是普通装饰器，它自己的元类重写了 __call__（crew_base.py:762）：

# project/crew_base.py:762
class _CrewBaseType(type):
    """Metaclass for CrewBase that makes it callable as a decorator."""
    def __call__(cls, decorated_cls: type) -> type[CrewClass]:
        __name  = str(decorated_cls.__name__)
        __bases = tuple(decorated_cls.__bases__)
        __dict  = {k: v for k, v in decorated_cls.__dict__.items()
                   if k not in ("__dict__", "__weakref__")}
        __dict["__metaclass__"] = CrewBaseMeta
        return cast(type[CrewClass], CrewBaseMeta(__name, __bases, __dict))  # 用新元类重造类

# :787
class CrewBase(metaclass=_CrewBaseType):
    """Class decorator that applies CrewBaseMeta metaclass."""

被换上的 CrewBaseMeta.__new__ 会在建类时注入配置与方法（crew_base.py:196）：

# project/crew_base.py:196
def __new__(mcs, name, bases, namespace, **kwargs):
    cls = cast(type[CrewClass], super().__new__(mcs, name, bases, namespace))
    cls.is_crew_class = True
    cls._crew_name = name
    for setup_fn in _CLASS_SETUP_FUNCTIONS:    # :221 设 base_directory / 配置路径 / MCP 参数
        setup_fn(cls)
    for method in _METHODS_TO_INJECT:          # :224 把 load_yaml/map_all_*等函数装进类
        setattr(cls, method.__name__, method)
    return cls

元类还重写了 __call__（实例化拦截），每次 MyCrew() 都会跑一遍初始化（crew_base.py:229）：

# project/crew_base.py:229
def __call__(cls, *args, **kwargs) -> CrewInstance:
    instance: CrewInstance = super().__call__(*args, **kwargs)
    CrewBaseMeta._initialize_crew_instance(instance, cls)   # 拦截：多做一步初始化
    return instance

# :243
@staticmethod
def _initialize_crew_instance(instance, cls):
    instance._mcp_server_adapter = None
    instance.load_configurations()          # ① 读两个 YAML → self.agents_config/tasks_config
    instance._all_methods = _get_all_methods(instance)
    instance.map_all_agent_variables()      # ② 把 YAML 里的字符串引用解析成实例（L05）
    instance.map_all_task_variables()
    ...
    original_methods = {name: m for name, m in cls.__dict__.items()
        if any(hasattr(m, attr) for attr in
               ["is_task","is_agent","is_before_kickoff","is_after_kickoff","is_kickoff"])}
    instance.__crew_metadata__ = CrewMetadata(         # ③ 按身份牌分类归档
        original_tasks   =_filter_methods(original_methods, "is_task"),
        original_agents  =_filter_methods(original_methods, "is_agent"),
        before_kickoff   =_filter_methods(original_methods, "is_before_kickoff"),
        after_kickoff    =after_kickoff_callbacks,
        kickoff          =_filter_methods(original_methods, "is_kickoff"))

而 load_configurations 就是读 YAML 文件（crew_base.py:368）：

# project/crew_base.py:368
def load_configurations(self):
    self.agents_config = self._load_config(self.original_agents_config_path, "agent")
    self.tasks_config  = self._load_config(self.original_tasks_config_path, "task")

# :378
def load_yaml(config_path: Path):
    with open(config_path, encoding="utf-8") as file:
        content = yaml.safe_load(file)
    return content if isinstance(content, dict) else {}

YAML 里 task 常写 agent: researcher（一个字符串）。解析发生在 _map_task_variables（crew_base.py:688）：

# project/crew_base.py:712
def _map_task_variables(self, task_name, task_info, agents, tasks, ...):
    if context_list := task_info.get("context"):        # context: [task_a, task_b]
        self.tasks_config[task_name]["context"] = [
            tasks[context_task_name]() for context_task_name in context_list]  # 名字→调工厂
    if tools := task_info.get("tools"):
        if _is_string_list(tools):                       # tools 全是字符串？
            self.tasks_config[task_name]["tools"] = [tool_functions[t]() for t in tools]
    if agent_name := task_info.get("agent"):             # agent: researcher
        self.tasks_config[task_name]["agent"] = agents[agent_name]()   # ★调 @agent 工厂拿实例
    if output_pydantic := task_info.get("output_pydantic"):
        self.tasks_config[task_name]["output_pydantic"] = output_pydantic_functions[output_pydantic]

Agent 侧同理，把 llm: my_llm、tools: [...] 等字符串解析成对象（crew_base.py:636）：

# project/crew_base.py:636
def _map_agent_variables(self, agent_name, agent_info, llms, tool_functions, ...):
    if llm := agent_info.get("llm"):
        factory = llms.get(llm)
        self.agents_config[agent_name]["llm"] = factory() if factory else llm  # 有@llm工厂就调，否则当模型名
    if tools := agent_info.get("tools"):
        if _is_string_list(tools):
            self.agents_config[agent_name]["tools"] = [tool_functions[tool]() for tool in tools]

@crew 是最"忙"的装饰器，它 wrapper 里替你把 self.agents/self.tasks 填好（annotations.py:200）：

# project/annotations.py:200
@wraps(meth)
def wrapper(self, *args, **kwargs) -> Crew:
    instantiated_tasks: list[Task] = []
    instantiated_agents: list[Agent] = []
    agent_roles: set[str] = set()

    tasks  = self.__crew_metadata__["original_tasks"].items()
    agents = self.__crew_metadata__["original_agents"].items()

    for _, task_method in tasks:                     # ① 先实例化所有 @task
        task_instance = _call_method(task_method, self)
        instantiated_tasks.append(task_instance)
        agent_instance = getattr(task_instance, "agent", None)
        if agent_instance and agent_instance.role not in agent_roles:  # 顺带收集任务用到的 agent
            instantiated_agents.append(agent_instance)
            agent_roles.add(agent_instance.role)

    for _, agent_method in agents:                   # ② 再补上剩余 @agent（去重）
        agent_instance = _call_method(agent_method, self)
        if agent_instance.role not in agent_roles:
            instantiated_agents.append(agent_instance)
            agent_roles.add(agent_instance.role)

    self.agents = instantiated_agents                # ③ 填好实例属性
    self.tasks  = instantiated_tasks
    crew_instance: Crew = _call_method(meth, self, *args, **kwargs)  # ④ 才真正调你写的 crew() 体
    ...
    for hook_callback in self.__crew_metadata__["before_kickoff"].values():   # ⑤ 挂 kickoff 钩子
        crew_instance.before_kickoff_callbacks.append(callback_wrapper(hook_callback, self))
    return crew_instance

注意 L02 里每个装饰器都 memoize(meth)。看它的实现（project/utils.py）：

# project/utils.py（memoize 精简）
def memoize(func):
    cache = {}
    @wraps(func)
    def wrapper(*args, **kwargs):
        key = str(args) + str(kwargs)
        if key not in cache:
            cache[key] = func(*args, **kwargs)   # 只在第一次真正执行
        return cache[key]                        # 之后都返回同一个对象
    return wrapper