Customizing the OTA Cycle¶

import os

model_name = os.environ.get("MODEL_NAME")
api_key = os.environ.get("API_KEY")
api_base = os.environ.get("BASE_URL")

from bridgic.llms.openai import OpenAILlm, OpenAIConfiguration

llm = OpenAILlm(
    api_key=api_key,
    api_base=api_base,
    timeout=30,
    configuration=OpenAIConfiguration(
        model=model_name,
        temperature=0.0,
        max_tokens=16384,
    ),
)

from bridgic.amphibious import OTAContext, Context, bash_tool

async def deploy_service(service: str) -> str:
    """Deploy a named service to production."""
    return f"deployed {service}"

async def delete_service(service: str) -> str:
    """Permanently delete a named service."""
    return f"deleted {service}"

# Small-loop context: declare the tools this run carries.
class OpsOTAContext(OTAContext):
    pass

OpsOTAContext.tool(deploy_service)
OpsOTAContext.tool(delete_service)
OpsOTAContext.tool(bash_tool)

# Big-loop context: free-form knowledge (bare here).
class OpsContext(Context):
    pass

from bridgic.amphibious import CognitiveWorker
from bridgic.core.model.types import Message, Role

class OpsThink(CognitiveWorker):
    async def thinking(self, ota_context, context=None):
        messages = [Message.from_text(ota_context.summary(), role=Role.USER)]
        # aselect_tool returns (tool_calls, content); a reply with no tool
        # calls is the finish.
        return await self._llm.aselect_tool(
            messages=messages,
            tools=[t.to_tool() for t in ota_context.tools],
        )

from bridgic.amphibious import _DELEGATE

class WorkerObsThink(OpsThink):
    async def observation(self, ota_context, context=None):
        # A returned value becomes this round's obs_result.
        if not ota_context.ota_record:
            return "Tip: deploy to staging before production."
        # Later rounds: chain to the agent-level observation.
        return _DELEGATE

from bridgic.amphibious import (
    AmphibiousAutoma, think_unit, ThinkUnit, RETURN,
)

class ObservingAgent(AmphibiousAutoma[OpsOTAContext, OpsContext]):
    worker = think_unit(WorkerObsThink(), max_attempts=5)

    # Agent-level observation: shared across all workers. The worker's
    # observation chains here when it returns _DELEGATE.
    async def observation(self, ota_context, context=None):
        yield RETURN("System: prod cluster, change window OPEN.")

    async def on_agent(self, ota_context, context=None):
        yield ThinkUnit("worker")

from bridgic.amphibious import ActionCall

class SnapshotAgent(AmphibiousAutoma[OpsOTAContext, OpsContext]):
    worker = think_unit(OpsThink(), max_attempts=5)

    async def observation(self, ota_context, context=None):
        # Live snapshot before each think: yield a raw tool call, then RETURN it.
        snapshot = yield ActionCall("bash", command="date -u")
        yield RETURN(snapshot[0].result if snapshot else None)

    async def on_agent(self, ota_context, context=None):
        yield ThinkUnit("worker")

class SafeAgent(AmphibiousAutoma[OpsOTAContext, OpsContext]):
    worker = think_unit(OpsThink(), max_attempts=5)

    BLOCKED = {"delete_service"}

    async def before_action(self, ota_context, context=None):
        # Payload-free: read the pending decision off the current round.
        decision = ota_context.think_result
        decision.tool_calls = [
            c for c in decision.tool_calls if c.tool not in self.BLOCKED
        ]
        yield RETURN(decision)  # override before the act phase runs

    async def on_agent(self, ota_context, context=None):
        yield ThinkUnit("worker")

class FilteringThink(OpsThink):
    BLOCKED = {"delete_service"}

    async def before_action(self, ota_context, context=None):
        decision = ota_context.think_result
        if any(c.tool in self.BLOCKED for c in decision.tool_calls):
            decision.tool_calls = [
                c for c in decision.tool_calls if c.tool not in self.BLOCKED
            ]
            return decision        # handled locally; stop the chain
        return _DELEGATE           # nothing to filter; chain to the agent

from bridgic.amphibious import ActionResult

class TrackingAgent(AmphibiousAutoma[OpsOTAContext, OpsContext]):
    worker = think_unit(OpsThink(), max_attempts=5)

    async def after_action(self, ota_context, context=None):
        action_result = ota_context.action_result
        if isinstance(action_result, ActionResult):
            succeeded = sum(1 for r in action_result.results if r.success)
            failed = sum(1 for r in action_result.results if not r.success)
            # Fold custom fields onto the current round (OTARecord is extra="allow").
            ota_context._current_record().succeeded = succeeded
            ota_context._current_record().failed = failed
        if False:  # keep this an async generator even when nothing yields
            yield

    async def on_agent(self, ota_context, context=None):
        yield ThinkUnit("worker")

class SequentialAgent(AmphibiousAutoma[OpsOTAContext, OpsContext]):
    worker = think_unit(OpsThink(), max_attempts=5)

    async def action_tool_call(self, ota_context, context=None) -> ActionResult:
        # A coroutine returning ActionResult. Add custom execution policy here
        # (e.g. a rate-limit gate), then delegate to the default executor.
        decision = ota_context.think_result
        if decision is not None and len(decision.tool_calls) > 3:
            # Example policy hook: cap fan-out before executing.
            decision.tool_calls = decision.tool_calls[:3]
        return await super().action_tool_call(ota_context, context)

    async def on_agent(self, ota_context, context=None):
        yield ThinkUnit("worker")