conditions

Class info

Classes

Name	Children	Inherits
AndCondition llmling_agent.models.conditions Require all conditions to be met.		ConnectionCondition
CallableCondition llmling_agent.models.conditions Custom predicate function.		ConnectionCondition
ConnectionCondition llmling_agent.models.conditions Base class for connection control conditions.	Jinja2Condition WordMatchCondition MessageCountCondition TimeCondition TokenThresholdCondition CostCondition CostLimitCondition CallableCondition AndCondition OrCondition	BaseModel
CostCondition llmling_agent.models.conditions Stop when cost threshold is reached.		ConnectionCondition
CostLimitCondition llmling_agent.models.conditions Disconnect when cost limit is reached.		ConnectionCondition
Jinja2Condition llmling_agent.models.conditions Evaluate condition using Jinja2 template.		ConnectionCondition
MessageCountCondition llmling_agent.models.conditions Disconnect after N messages.		ConnectionCondition
OrCondition llmling_agent.models.conditions Require any condition to be met.		ConnectionCondition
TimeCondition llmling_agent.models.conditions Disconnect after time period.		ConnectionCondition
TokenThresholdCondition llmling_agent.models.conditions Disconnect after token threshold is reached.		ConnectionCondition
WordMatchCondition llmling_agent.models.conditions Disconnect when word/phrase is found in message.		ConnectionCondition

🛈 DocStrings

AndCondition

Bases: ConnectionCondition

Require all conditions to be met.

Source code in src/llmling_agent/models/conditions.py

class AndCondition(ConnectionCondition):
    """Require all conditions to be met."""

    type: Literal["and"] = Field("and", init=False)
    """Condition to AND-combine multiple conditions."""

    conditions: list[ConnectionCondition]
    """List of conditions to check."""

    async def check(self, context: EventContext) -> bool:
        """Check if all conditions are met."""
        results = [await c.check(context) for c in self.conditions]
        return all(results)

conditions instance-attribute

conditions: list[ConnectionCondition]

List of conditions to check.

type class-attribute instance-attribute

type: Literal['and'] = Field('and', init=False)

Condition to AND-combine multiple conditions.

check async

check(context: EventContext) -> bool

Check if all conditions are met.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if all conditions are met."""
    results = [await c.check(context) for c in self.conditions]
    return all(results)

CallableCondition

Bases: ConnectionCondition

Custom predicate function.

Source code in src/llmling_agent/models/conditions.py

class CallableCondition(ConnectionCondition):
    """Custom predicate function."""

    type: Literal["callable"] = Field("callable", init=False)
    """Condition based on an import path pointing to a predicate."""

    predicate: ImportString[Callable[..., bool | Awaitable[bool]]]
    """Function to evaluate condition:
    Args:
        message: Current message being processed
        stats: Current connection statistics
    Returns:
        Whether condition is met
    """

    async def check(self, context: EventContext) -> bool:
        """Execute predicate function."""
        return await execute(self.predicate, context.message, context.stats)

predicate instance-attribute

predicate: ImportString[Callable[..., bool | Awaitable[bool]]]

Function to evaluate condition: Args: message: Current message being processed stats: Current connection statistics Returns: Whether condition is met

type class-attribute instance-attribute

type: Literal['callable'] = Field('callable', init=False)

Condition based on an import path pointing to a predicate.

check async

check(context: EventContext) -> bool

Execute predicate function.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Execute predicate function."""
    return await execute(self.predicate, context.message, context.stats)

ConnectionCondition

Bases: BaseModel

Base class for connection control conditions.

Source code in src/llmling_agent/models/conditions.py

class ConnectionCondition(BaseModel):
    """Base class for connection control conditions."""

    type: str = Field(init=False)
    """Discriminator for condition types."""

    name: str | None = None
    """Optional name for the condition for referencing."""

    model_config = ConfigDict(frozen=True, use_attribute_docstrings=True, extra="forbid")

    async def check(self, context: EventContext) -> bool:
        """Check if condition is met."""
        raise NotImplementedError

name class-attribute instance-attribute

name: str | None = None

Optional name for the condition for referencing.

type class-attribute instance-attribute

type: str = Field(init=False)

Discriminator for condition types.

check async

check(context: EventContext) -> bool

Check if condition is met.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if condition is met."""
    raise NotImplementedError

CostCondition

Bases: ConnectionCondition

Stop when cost threshold is reached.

Source code in src/llmling_agent/models/conditions.py

class CostCondition(ConnectionCondition):
    """Stop when cost threshold is reached."""

    type: Literal["cost"] = Field("cost", init=False)
    """Cost-based condition."""

    max_cost: float
    """Maximum cost in USD."""

    async def check(self, context: EventContext) -> bool:
        """Check if cost limit is reached."""
        return context.stats.total_cost >= self.max_cost

max_cost instance-attribute

max_cost: float

Maximum cost in USD.

type class-attribute instance-attribute

type: Literal['cost'] = Field('cost', init=False)

Cost-based condition.

check async

check(context: EventContext) -> bool

Check if cost limit is reached.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if cost limit is reached."""
    return context.stats.total_cost >= self.max_cost

CostLimitCondition

Bases: ConnectionCondition

Disconnect when cost limit is reached.

Source code in src/llmling_agent/models/conditions.py

class CostLimitCondition(ConnectionCondition):
    """Disconnect when cost limit is reached."""

    type: Literal["cost_limit"] = Field("cost_limit", init=False)
    """Cost-limit condition."""

    max_cost: float
    """Maximum cost in USD before triggering."""

    async def check(self, context: EventContext) -> bool:
        """Check if cost limit is reached."""
        if not context.message.cost_info:
            return False
        return float(context.message.cost_info.total_cost) >= self.max_cost

max_cost instance-attribute

max_cost: float

Maximum cost in USD before triggering.

type class-attribute instance-attribute

type: Literal['cost_limit'] = Field('cost_limit', init=False)

Cost-limit condition.

check async

check(context: EventContext) -> bool

Check if cost limit is reached.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if cost limit is reached."""
    if not context.message.cost_info:
        return False
    return float(context.message.cost_info.total_cost) >= self.max_cost

Jinja2Condition

Bases: ConnectionCondition

Evaluate condition using Jinja2 template.

Source code in src/llmling_agent/models/conditions.py

class Jinja2Condition(ConnectionCondition):
    """Evaluate condition using Jinja2 template."""

    type: Literal["jinja2"] = Field("jinja2", init=False)
    """Jinja2 template-based condition."""

    template: str
    """Jinja2 template to evaluate."""

    async def check(self, ctx: EventContext) -> bool:
        from datetime import datetime

        from jinja2 import Environment

        env = Environment(trim_blocks=True, lstrip_blocks=True)
        template = env.from_string(self.template)

        # Just pass the entire context to the template
        result = template.render(
            ctx=ctx,
            now=datetime.now(),  # Extra convenience
        )

        return result.strip().lower() == "true" or bool(result)

template instance-attribute

template: str

Jinja2 template to evaluate.

type class-attribute instance-attribute

type: Literal['jinja2'] = Field('jinja2', init=False)

Jinja2 template-based condition.

MessageCountCondition

Bases: ConnectionCondition

Disconnect after N messages.

Source code in src/llmling_agent/models/conditions.py

class MessageCountCondition(ConnectionCondition):
    """Disconnect after N messages."""

    type: Literal["message_count"] = Field("message_count", init=False)
    """Message-count-based condition."""

    max_messages: int
    """Maximum number of messages before triggering."""

    count_mode: Literal["total", "per_agent"] = "total"
    """How to count messages:
    - total: All messages in conversation
    - per_agent: Messages from each agent separately
    """

    async def check(self, context: EventContext) -> bool:
        """Check if message count threshold is reached."""
        if self.count_mode == "total":
            return context.stats.message_count >= self.max_messages

        # Count per agent
        agent_messages = [
            m for m in context.stats.messages if m.name == context.message.name
        ]
        return len(agent_messages) >= self.max_messages

count_mode class-attribute instance-attribute

count_mode: Literal['total', 'per_agent'] = 'total'

How to count messages: - total: All messages in conversation - per_agent: Messages from each agent separately

max_messages instance-attribute

max_messages: int

Maximum number of messages before triggering.

type class-attribute instance-attribute

type: Literal['message_count'] = Field('message_count', init=False)

Message-count-based condition.

check async

check(context: EventContext) -> bool

Check if message count threshold is reached.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if message count threshold is reached."""
    if self.count_mode == "total":
        return context.stats.message_count >= self.max_messages

    # Count per agent
    agent_messages = [
        m for m in context.stats.messages if m.name == context.message.name
    ]
    return len(agent_messages) >= self.max_messages

OrCondition

Bases: ConnectionCondition

Require any condition to be met.

Source code in src/llmling_agent/models/conditions.py

class OrCondition(ConnectionCondition):
    """Require any condition to be met."""

    type: Literal["or"] = Field("or", init=False)
    """Condition to OR-combine multiple conditions."""

    conditions: list[ConnectionCondition]
    """List of conditions to check."""

    async def check(self, context: EventContext) -> bool:
        """Check if any condition is met."""
        results = [await c.check(context) for c in self.conditions]
        return any(results)

conditions instance-attribute

conditions: list[ConnectionCondition]

List of conditions to check.

type class-attribute instance-attribute

type: Literal['or'] = Field('or', init=False)

Condition to OR-combine multiple conditions.

check async

check(context: EventContext) -> bool

Check if any condition is met.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if any condition is met."""
    results = [await c.check(context) for c in self.conditions]
    return any(results)

TimeCondition

Bases: ConnectionCondition

Disconnect after time period.

Source code in src/llmling_agent/models/conditions.py

class TimeCondition(ConnectionCondition):
    """Disconnect after time period."""

    type: Literal["time"] = Field("time", init=False)
    """Time-based condition."""

    duration: timedelta
    """How long the connection should stay active."""

    async def check(self, context: EventContext) -> bool:
        """Check if time duration has elapsed."""
        elapsed = datetime.now() - context.stats.start_time
        return elapsed >= self.duration

duration instance-attribute

duration: timedelta

How long the connection should stay active.

type class-attribute instance-attribute

type: Literal['time'] = Field('time', init=False)

Time-based condition.

check async

check(context: EventContext) -> bool

Check if time duration has elapsed.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if time duration has elapsed."""
    elapsed = datetime.now() - context.stats.start_time
    return elapsed >= self.duration

TokenThresholdCondition

Bases: ConnectionCondition

Disconnect after token threshold is reached.

Source code in src/llmling_agent/models/conditions.py

class TokenThresholdCondition(ConnectionCondition):
    """Disconnect after token threshold is reached."""

    type: Literal["token_threshold"] = Field("token_threshold", init=False)
    """Type discriminator."""

    max_tokens: int
    """Maximum number of tokens allowed."""

    count_type: Literal["total", "prompt", "completion"] = "total"
    """What tokens to count:
    - total: All tokens used
    - prompt: Only prompt tokens
    - completion: Only completion tokens
    """

    async def check(self, context: EventContext) -> bool:
        """Check if token threshold is reached."""
        if not context.message.cost_info:
            return False

        match self.count_type:
            case "total":
                return context.stats.token_count >= self.max_tokens
            case "prompt":
                return context.message.cost_info.token_usage["prompt"] >= self.max_tokens
            case "completion":
                return (
                    context.message.cost_info.token_usage["completion"] >= self.max_tokens
                )

count_type class-attribute instance-attribute

count_type: Literal['total', 'prompt', 'completion'] = 'total'

What tokens to count: - total: All tokens used - prompt: Only prompt tokens - completion: Only completion tokens

max_tokens instance-attribute

max_tokens: int

Maximum number of tokens allowed.

type class-attribute instance-attribute

type: Literal['token_threshold'] = Field('token_threshold', init=False)

Type discriminator.

check async

check(context: EventContext) -> bool

Check if token threshold is reached.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if token threshold is reached."""
    if not context.message.cost_info:
        return False

    match self.count_type:
        case "total":
            return context.stats.token_count >= self.max_tokens
        case "prompt":
            return context.message.cost_info.token_usage["prompt"] >= self.max_tokens
        case "completion":
            return (
                context.message.cost_info.token_usage["completion"] >= self.max_tokens
            )

WordMatchCondition

Bases: ConnectionCondition

Disconnect when word/phrase is found in message.

Source code in src/llmling_agent/models/conditions.py

class WordMatchCondition(ConnectionCondition):
    """Disconnect when word/phrase is found in message."""

    type: Literal["word_match"] = Field("word_match", init=False)
    """Word-comparison-based condition."""

    words: list[str]
    """Words or phrases to match in messages."""

    case_sensitive: bool = False
    """Whether to match case-sensitively."""

    mode: Literal["any", "all"] = "any"
    """Match mode:
    - any: Trigger if any word matches
    - all: Require all words to match
    """

    async def check(self, context: EventContext) -> bool:
        """Check if message contains specified words."""
        text = str(context.message.content)
        if not self.case_sensitive:
            text = text.lower()
            words = [w.lower() for w in self.words]
        else:
            words = self.words

        matches = [w in text for w in words]
        return all(matches) if self.mode == "all" else any(matches)

case_sensitive class-attribute instance-attribute

case_sensitive: bool = False

Whether to match case-sensitively.

mode class-attribute instance-attribute

mode: Literal['any', 'all'] = 'any'

Match mode: - any: Trigger if any word matches - all: Require all words to match

type class-attribute instance-attribute

type: Literal['word_match'] = Field('word_match', init=False)

Word-comparison-based condition.

words instance-attribute

words: list[str]

Words or phrases to match in messages.

check async

check(context: EventContext) -> bool

Check if message contains specified words.

Source code in src/llmling_agent/models/conditions.py

async def check(self, context: EventContext) -> bool:
    """Check if message contains specified words."""
    text = str(context.message.content)
    if not self.case_sensitive:
        text = text.lower()
        words = [w.lower() for w in self.words]
    else:
        words = self.words

    matches = [w in text for w in words]
    return all(matches) if self.mode == "all" else any(matches)