资源（Resources）是可以注入到工作流步骤中的外部依赖。

凡是应当由运行时来创建、而不是当事件传递或存进 ctx.store 的对象，都用资源来表达：LLM 客户端、检索器、索引、数据库句柄、模型配置，以及其他开销大、有状态或无法 JSON 序列化的依赖。

举个简单例子，看看下面这个工作流里来自 LlamaIndex 的 Memory：

from typing import Annotated

from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent
from workflows.resource import Resource
from llama_index.core.llms import ChatMessage
from llama_index.core.memory import Memory

def get_memory() -> Memory:
    return Memory.from_defaults("user_id_123", token_limit=60000)

class SecondEvent(Event):
    msg: str

class WorkflowWithResource(Workflow):
    @step
    async def first_step(
        self,
        ev: StartEvent,
        memory: Annotated[Memory, Resource(get_memory)],
    ) -> SecondEvent:
        print("Memory before step 1", memory)
        await memory.aput(
            ChatMessage(role="user", content="This is the first step")
        )
        print("Memory after step 1", memory)
        return SecondEvent(msg="This is an input for step 2")

    @step
    async def second_step(
        self, ev: SecondEvent, memory: Annotated[Memory, Resource(get_memory)]
    ) -> StopEvent:
        print("Memory before step 2", memory)
        await memory.aput(ChatMessage(role="user", content=ev.msg))
        print("Memory after step 2", memory)
        return StopEvent(result="Messages put into memory")

要注入一个资源，就在步骤签名里加一个参数、把类型包进 Annotated，并给 Resource() 传一个工厂：

memory: Annotated[Memory, Resource(get_memory)]

工厂的返回类型应与被标注的参数类型一致。默认情况下，资源会在本次工作流运行内缓存，因此两个步骤拿到的是同一个 Memory 对象、工厂只调用一次。如果某个步骤每次都需要新对象，请传入 cache=False：

memory: Annotated[Memory, Resource(get_memory, cache=False)]

工厂可以是同步的，也可以是异步的。

由配置驱动的资源

对于存放在 JSON 文件里的配置数据，请用 ResourceConfig 而非 Resource。它会加载 JSON 文件并解析成一个 Pydantic 模型。

from typing import Annotated
from pydantic import BaseModel
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.resource import ResourceConfig

class ClassifierConfig(BaseModel):
    categories: list[str]
    threshold: float

class DocumentClassifier(Workflow):
    @step
    async def classify(
        self,
        ev: StartEvent,
        config: Annotated[
            ClassifierConfig,
            ResourceConfig(config_file="classifier.json"),
        ],
    ) -> StopEvent:
        # config is loaded from classifier.json and validated as ClassifierConfig
        return StopEvent(result=f"Using threshold: {config.threshold}")

参数

• config_file：配置文件（JSON）的路径。
• path_selector：可选的、以 ”.” 分隔的 JSON 路径，用于从 JSON 文件中取出嵌套值（例如 "settings.classifier"）。
• label：可选的显示名，用于工作流可视化。
• description：可选的描述，用于工作流可视化。

选取嵌套值

如果你的 JSON 文件里含多份配置，用 path_selector 取出特定的那一节：

# Given config.json: {"classifier": {"categories": [...], "threshold": 0.8}, "other": {...}}
config: Annotated[
    ClassifierConfig,
    ResourceConfig(config_file="config.json", path_selector="classifier"),
]

可视化中的标签与描述

在调试器或其他可视化工具中查看工作流时，label 与 description 有助于辨识各个配置：

config: Annotated[
    ClassifierConfig,
    ResourceConfig(
        config_file="classifier.json",
        label="Document Classifier",
        description="Categories and confidence threshold for classification",
    ),
]

如果没有提供标签，就使用 Pydantic 模型的类型名（例如 “ClassifierConfig”）。

资源链式依赖

Resource 与 ResourceConfig 可以串联起来。Resource 的工厂函数可以用同样的 Annotated 写法声明对其他资源的依赖：

from typing import Annotated
from pydantic import BaseModel
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.resource import Resource, ResourceConfig
from llama_index.llms.anthropic import Anthropic

class LLMConfig(BaseModel):
    model: str
    temperature: float
    max_tokens: int

def get_llm(
    config: Annotated[LLMConfig, ResourceConfig(config_file="llm.json")],
) -> Anthropic:
    return Anthropic(
        model=config.model,
        temperature=config.temperature,
        max_tokens=config.max_tokens,
    )

class MyWorkflow(Workflow):
    @step
    async def generate(
        self,
        ev: StartEvent,
        llm: Annotated[Anthropic, Resource(get_llm)],
    ) -> StopEvent:
        response = await llm.acomplete(ev.input)
        return StopEvent(result=response.text)

依赖链会自动解析。在这个例子中，当工作流运行时：

1. 加载 llm.json 并解析成 LLMConfig
2. 用该配置调用 get_llm 创建 LLM 客户端
3. 把得到的客户端传给步骤

这种写法对 Resource 与 ResourceConfig 的任意组合依赖都适用。