bitsky-tech/AmphiLoop

Bridgic Amphibious

Framework de agente dual-mode: agentes operam em modo orientado por LLM (on_agent) e modo determinístico (on_workflow) com fallback automático entre eles.

Dependências

Um projeto bridgic-amphibious requer os seguintes pacotes:

Pacote	Descrição
bridgic-core	Framework principal (Worker, Automa, GraphAutoma, ASL)
bridgic-amphibious	Framework de agente dual-mode
bridgic-llms-openai	Provedor LLM (necessário apenas para modos AGENT / AMPHIFLOW)
python-dotenv	Carregamento de arquivos .env

Antes de usar este pacote, você precisa instalar as dependências usando o script de instalação fornecido:

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

O script verifica a disponibilidade do uv, inicializa um projeto uv se necessário, instala quaisquer pacotes ausentes via uv add, e executa uv sync para finalizar o ambiente. Quando ele sai com sucesso, o projeto está totalmente inicializado e pronto para uso — nenhum acompanhamento manual de uv add / uv sync é necessário.

Configuração do LLM

Agentes Amphibious aceitam uma instância BaseLlm com o protocolo astructure_output de um pacote provedor LLM bridgic. O LLM é necessário para os modos AGENT e AMPHIFLOW; o modo puro WORKFLOW pode ser executado sem um.

from bridgic.llms.openai import OpenAILlm, OpenAIConfiguration

llm = OpenAILlm(
    api_key="sua-chave-api",
    api_base="https://api.openai.com/v1",  # ou endpoint personalizado
    configuration=OpenAIConfiguration(model="gpt-4o", temperature=0.0),
)

Outros provedores com o mesmo protocolo: bridgic.llms.vllm.VllmServerLlm (vLLM auto-hospedado).

Início Rápido

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

async def get_weather(city: str) -> str:
    """Obtém a previsão do tempo para uma cidade."""
    return f"Ensolarado, 22°C em {city}"

class WeatherAgent(AmphibiousAutoma[CognitiveContext]):
    planner = think_unit(
        CognitiveWorker.inline("Verifique a previsão do tempo e forneça um resumo."),
        max_attempts=5,
    )
    async def on_agent(self, ctx: CognitiveContext):
        await self.planner

agent = WeatherAgent(llm=llm, verbose=True)
result = await agent.arun(
    goal="Verifique a previsão do tempo em Tóquio e Londres.",
    tools=[FunctionToolSpec.from_raw(get_weather)],
)

Scaffolding de Projeto

Use a CLI para inicializar um novo projeto:

bridgic-amphibious create -n meu_projeto
bridgic-amphibious create -n meu_projeto --task "Navegue até example.com e extraia dados"
bridgic-amphibious create -n meu_projeto --base-dir /caminho/para/projetos

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

Conceitos Principais

Agente = Unidades de Pensamento + Orquestração de Contexto. Agentes são definidos declarando unidades de pensamento CognitiveWorker e orquestrando-as em on_agent() ou on_workflow().

Arquitetura de quatro camadas:

1. Exposure — abstração de visibilidade de dados (LayeredExposure / EntireExposure)
2. CognitiveContext — contêiner de estado (goal, tools, skills, history)
3. CognitiveWorker — unidade de pensamento pura (observar-pensar-agir)
4. AmphibiousAutoma — motor de orquestração (roteamento de modo, ciclo de vida)

Ciclo OTC: Observar -> Pensar -> Agir, com pontos de gancho em cada fase.

Quatro RunModes: AGENT (orientado por LLM), WORKFLOW (determinístico), AMPHIFLOW (workflow + fallback de agente), AUTO (detecção automática a partir de métodos sobrescritos, padrão).

Resolução AUTO: apenas on_agent sobrescrito → AGENT; apenas on_workflow sobrescrito → WORKFLOW; ambos sobrescritos → AMPHIFLOW.

Padrões Principais

Modo Agente — LLM decide

class MyAgent(AmphibiousAutoma[CognitiveContext]):
    worker = think_unit(CognitiveWorker.inline("Decida o próximo passo."), max_attempts=10)
    async def on_agent(self, ctx):
        await self.worker

Modo Workflow — Desenvolvedor decide

from bridgic.amphibious import ActionCall

class MyWorkflow(AmphibiousAutoma[CognitiveContext]):
    async def on_workflow(self, ctx):
        result = yield ActionCall("nome_ferramenta", arg1="valor")
        # result é List[ToolResult]

# Modo workflow puro não precisa de um LLM.
await MyWorkflow().arun(goal="...", tools=[...])

Modo Amphiflow — Workflow com fallback de agente

from bridgic.amphibious import RunMode, AgentCall

class MyHybrid(AmphibiousAutoma[CognitiveContext]):
    fixer = think_unit(CognitiveWorker.inline("Corrija o problema."), max_attempts=5)
    async def on_agent(self, ctx): await self.fixer
    async def on_workflow(self, ctx):
        yield ActionCall("fill_field", name="user", value="john")
        yield ActionCall("click_button", name="submit")

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

Humano-no-loop

from bridgic.amphibious import ActionCall, HumanCall

class MyAgent(AmphibiousAutoma[CognitiveContext]):
    worker = think_unit(CognitiveWorker.inline("Execute o passo."), max_attempts=10)

    async def on_agent(self, ctx):
        await self.worker
        feedback = await self.request_human("Prosseguir?")  # Ponto 1: nível de código

    async def on_workflow(self, ctx):
        yield ActionCall("do_something", arg="valor")
        feedback = yield HumanCall(prompt="Confirmar?")     # Ponto 2: yield do workflow

# Ponto 3: ferramenta LLM — `request_human` é injetada automaticamente nas ferramentas
# de todo agente, então o LLM pode chamá-la sem que você a liste em tools=[...].
# (Passar `request_human_tool` explicitamente é inofensivo — é deduplicado.)
await MyAgent(llm=llm).arun(goal="...", tools=[minha_ferramenta])

# Se você quiser ser explícito, importar e passar `request_human_tool` ainda funciona.
from bridgic.amphibious.builtin_tools import request_human_tool
await agent.arun(goal="...", tools=[request_human_tool, ...])  # também funciona

Saída Pydantic Personalizada

from pydantic import BaseModel

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

class Planner(AmphibiousAutoma[CognitiveContext]):
    plan = think_unit(
        CognitiveWorker.inline("Crie um plano.", output_schema=Plan),
        max_attempts=1,
    )
    async def on_agent(self, ctx):
        result = await self.plan  # Retorna instância de Plan

Anotação de Fase (snapshot)

async def on_agent(self, ctx):
    async with self.snapshot(goal="Fase de pesquisa"):
        await self.researcher
    async with self.snapshot(goal="Fase de escrita"):
        await self.writer

Arquivos de Referência

• Detalhes da arquitetura (modos de execução, sistema de exposição, níveis de memória, políticas cognitivas): Veja references/architecture.md
• Referência completa da API (todas as classes, métodos, parâmetros, tipos): Veja references/api-reference.md
• Padrões de código e exemplos completos (todos os tipos de gancho, skills, rastreamento, filtragem, etc.): Veja references/patterns.md