Skip to content

context

Class info

Classes

Name Children Inherits
NodeContext
llmling_agent.messaging.context
NodeContext(*, node_name: 'str', pool: 'AgentPool | None' \= None, config: 'NodeConfig', definition: 'AgentsManifest', current_prompt: 'str | None' \= None, in_async_context: 'bool' \= False, input_provider: 'InputProvider | None' \= None)

🛈 DocStrings

Base class for message processing nodes.

NodeContext dataclass

Source code in src/llmling_agent/messaging/context.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@dataclass(kw_only=True)
class NodeContext[TDeps]:
    node_name: str
    """Name of the current node."""

    pool: AgentPool | None = None
    """The agent pool the node is part of."""

    config: NodeConfig
    """Node configuration."""

    definition: AgentsManifest
    """Complete agent definition with all configurations."""

    current_prompt: str | None = None
    """Current prompt text for the agent."""

    in_async_context: bool = False
    """Whether we're running in an async context."""

    input_provider: InputProvider | None = None
    """Provider for human-input-handling."""

    def get_input_provider(self) -> InputProvider:
        from llmling_agent_input.stdlib_provider import StdlibInputProvider

        if self.input_provider:
            return self.input_provider
        if self.pool and self.pool._input_provider:
            return self.pool._input_provider
        return StdlibInputProvider()

    @property
    def node(self) -> MessageEmitter[TDeps, Any]:
        """Get the agent instance from the pool."""
        assert self.pool, "No agent pool available"
        assert self.node_name, "No agent name available"
        return self.pool[self.node_name]  # pyright: ignore

    @cached_property
    def storage(self) -> StorageManager:
        """Storage manager from pool or config."""
        from llmling_agent.storage import StorageManager

        if self.pool:
            return self.pool.storage
        return StorageManager(self.definition.storage)

    @property
    def prompt_manager(self) -> PromptManager:
        """Get prompt manager from manifest."""
        return self.definition.prompt_manager

    @property
    def report_progress(self) -> ProgressCallback | None:
        """Access progress reporting from pool server if available."""
        return (
            self.pool.server.report_progress if self.pool and self.pool.server else None
        )

config instance-attribute 

config: NodeConfig

Node configuration.

current_prompt class-attribute instance-attribute 

current_prompt: str | None = None

Current prompt text for the agent.

definition instance-attribute 

definition: AgentsManifest

Complete agent definition with all configurations.

in_async_context class-attribute instance-attribute 

in_async_context: bool = False

Whether we're running in an async context.

input_provider class-attribute instance-attribute 

input_provider: InputProvider | None = None

Provider for human-input-handling.

node property 

node: MessageEmitter[TDeps, Any]

Get the agent instance from the pool.

node_name instance-attribute 

node_name: str

Name of the current node.

pool class-attribute instance-attribute 

pool: AgentPool | None = None

The agent pool the node is part of.

prompt_manager property 

prompt_manager: PromptManager

Get prompt manager from manifest.

report_progress property 

report_progress: ProgressCallback | None

Access progress reporting from pool server if available.

storage cached property 

storage: StorageManager

Storage manager from pool or config.