bitsky-tech/AmphiLoop

Bridgic Amphibious

Zwei-Modus-Agenten-Framework: Agenten arbeiten in LLM-gesteuertem (on_agent) und deterministischem (on_workflow) Modus mit automatischem Fallback zwischen ihnen.

Abhängigkeiten

Ein Bridgic-Amphibious-Projekt benötigt die folgenden Pakete:

Paket	Beschreibung
bridgic-core	Kern-Framework (Worker, Automa, GraphAutoma, ASL)
bridgic-amphibious	Zwei-Modus-Agenten-Framework
bridgic-llms-openai	LLM-Anbieter (nur für AGENT / AMPHIFLOW Modi erforderlich)
python-dotenv	.env-Datei laden

Bevor Sie dieses Paket verwenden, müssen Sie die Abhängigkeiten mit dem bereitgestellten Installationsskript installieren:

bash "skills/bridgic-amphibious/scripts/install-deps.sh" "$PWD"

Das Skript prüft die Verfügbarkeit von uv, initialisiert bei Bedarf ein uv-Projekt, installiert fehlende Pakete über uv add und führt uv sync aus, um die Umgebung abzuschließen. Wenn es erfolgreich beendet wird, ist das Projekt vollständig initialisiert und einsatzbereit – kein manuelles uv add / uv sync ist erforderlich.

LLM-Einrichtung

Amphibische Agenten akzeptieren eine BaseLlm-Instanz mit astructure_output-Protokoll aus einem Bridgic-LLM-Anbieterpaket. Das LLM ist für die Modi AGENT und AMPHIFLOW erforderlich; der reine WORKFLOW-Modus kann ohne LLM ausgeführt werden.

from bridgic.llms.openai import OpenAILlm, OpenAIConfiguration

llm = OpenAILlm(
    api_key="ihr-api-key",
    api_base="https://api.openai.com/v1",  # oder benutzerdefinierter Endpunkt
    configuration=OpenAIConfiguration(model="gpt-4o", temperature=0.0),
)

Andere Anbieter mit demselben Protokoll: bridgic.llms.vllm.VllmServerLlm (selbst gehostetes vLLM).

Kurzanleitung

from bridgic.amphibious import (
    AmphibiousAutoma, CognitiveContext, CognitiveWorker, think_unit,
)
from bridgic.core.agentic.tool_specs import FunctionToolSpec

async def get_weather(city: str) -> str:
    """Wetter für eine Stadt abrufen."""
    return f"Sonnig, 22°C in {city}"

class WeatherAgent(AmphibiousAutoma[CognitiveContext]):
    planer = think_unit(
        CognitiveWorker.inline("Wetter nachschlagen und Zusammenfassung erstellen."),
        max_attempts=5,
    )
    async def on_agent(self, ctx: CognitiveContext):
        await self.planer

agent = WeatherAgent(llm=llm, verbose=True)
ergebnis = await agent.arun(
    ziel="Überprüfe das Wetter in Tokio und London.",
    tools=[FunctionToolSpec.from_raw(get_weather)],
)

Projektgerüst

Verwenden Sie die CLI, um ein neues Projekt zu bootstrappen:

bridgic-amphibious create -n mein_projekt
bridgic-amphibious create -n mein_projekt --task "Navigiere zu example.com und extrahiere Daten"
bridgic-amphibious create -n mein_projekt --base-dir /pfad/zu/projekten

Erstellt: task.md, config.py, tools.py, workers.py, agents.py, skills/, result/, log/.

Kernkonzepte

Agent = Denkeinheiten + Kontext-Orchestrierung. Agenten werden definiert, indem CognitiveWorker-Denkeinheiten deklariert und in on_agent() oder on_workflow() orchestriert werden.

Vierschichtige Architektur:

1. Exposure — Daten-Sichtbarkeitsabstraktion (LayeredExposure / EntireExposure)
2. CognitiveContext — Zustandscontainer (Ziel, Tools, Fähigkeiten, Verlauf)
3. CognitiveWorker — reine Denkeinheit (beobachten-denken-handeln)
4. AmphibiousAutoma — Orchestrierungs-Engine (Modus-Routing, Lebenszyklus)

OTC-Zyklus: Beobachten -> Denken -> Handeln, mit Hook-Punkten in jeder Phase.

Vier RunModes: AGENT (LLM-gesteuert), WORKFLOW (deterministisch), AMPHIFLOW (Workflow + Agent-Fallback), AUTO (automatische Erkennung basierend auf überschriebenen Methoden, Standard).

AUTO-Auflösung: nur on_agent überschrieben → AGENT; nur on_workflow überschrieben → WORKFLOW; beide überschrieben → AMPHIFLOW.

Wichtige Muster

Agenten-Modus – LLM entscheidet

class MyAgent(AmphibiousAutoma[CognitiveContext]):
    arbeiter = think_unit(CognitiveWorker.inline("Nächsten Schritt entscheiden."), max_attempts=10)
    async def on_agent(self, ctx):
        await self.arbeiter

Workflow-Modus – Entwickler entscheidet

from bridgic.amphibious import ActionCall

class MyWorkflow(AmphibiousAutoma[CognitiveContext]):
    async def on_workflow(self, ctx):
        ergebnis = yield ActionCall("tool_name", arg1="wert")
        # ergebnis ist List[ToolResult]

# Der reine Workflow-Modus benötigt kein LLM.
await MyWorkflow().arun(ziel="...", tools=[...])

Amphiflow-Modus – Workflow mit Agent-Fallback

from bridgic.amphibious import RunMode, AgentCall

class MyHybrid(AmphibiousAutoma[CognitiveContext]):
    reparierer = think_unit(CognitiveWorker.inline("Das Problem beheben."), max_attempts=5)
    async def on_agent(self, ctx): await self.reparierer
    async def on_workflow(self, ctx):
        yield ActionCall("fill_field", name="benutzer", value="john")
        yield ActionCall("click_button", name="submit")

await MyHybrid(llm=llm).arun(
    ziel="...", tools=[...],
    mode=RunMode.AMPHIFLOW, max_consecutive_fallbacks=2,
)

Human-in-the-Loop

from bridgic.amphibious import ActionCall, HumanCall

class MyAgent(AmphibiousAutoma[CognitiveContext]):
    arbeiter = think_unit(CognitiveWorker.inline("Schritt ausführen."), max_attempts=10)

    async def on_agent(self, ctx):
        await self.arbeiter
        feedback = await self.request_human("Fortfahren?")  # Einstieg 1: Code-Ebene

    async def on_workflow(self, ctx):
        yield ActionCall("do_something", arg="wert")
        feedback = yield HumanCall(prompt="Bestätigen?")     # Einstieg 2: Workflow-Yield

# Einstieg 3: LLM-Tool – `request_human` wird automatisch in die Tools jedes
# Agenten eingefügt, sodass das LLM es aufrufen kann, ohne dass Sie es in tools=[...] auflisten müssen.
# (Explizites Übergeben von `request_human_tool` ist harmlos – es wird dedupliziert.)
await MyAgent(llm=llm).arun(ziel="...", tools=[my_tool])

# Wenn Sie explizit sein möchten, ist das Importieren und Übergeben von `request_human_tool` weiterhin in Ordnung.
from bridgic.amphibious.builtin_tools import request_human_tool
await agent.arun(ziel="...", tools=[request_human_tool, ...])  # auch in Ordnung

Benutzerdefinierte Pydantic-Ausgabe

from pydantic import BaseModel

class Plan(BaseModel):
    phasen: list[str]

class Planner(AmphibiousAutoma[CognitiveContext]):
    plan = think_unit(
        CognitiveWorker.inline("Einen Plan erstellen.", output_schema=Plan),
        max_attempts=1,
    )
    async def on_agent(self, ctx):
        ergebnis = await self.plan  # Gibt Plan-Instanz zurück

Phasenannotation (Snapshot)

async def on_agent(self, ctx):
    async with self.snapshot(ziel="Recherchephase"):
        await self.forscher
    async with self.snapshot(ziel="Schreibphase"):
        await self.schreiber

Referenzdateien

• Architekturdetails (Ausführungsmodi, Exposure-System, Speicherstufen, kognitive Richtlinien): Siehe references/architecture.md
• Vollständige API-Referenz (alle Klassen, Methoden, Parameter, Typen): Siehe references/api-reference.md
• Vollständige Code-Muster und Beispiele (alle Hook-Typen, Fähigkeiten, Tracing, Filterung usw.): Siehe references/patterns.md