Show HN: Druids – 자신만의 소프트웨어 팩토리 구축하기

ramure는 신뢰할 수 있는 에이전트 소프트웨어 (agent software)를 구축하기 위한 주관이 뚜렷하고 가벼운 Python 라이브러리입니다. 이 라이브러리는 에이전트들이 작업을 완수하기 위해 환경 간에 통신하는 프로그램을 쉽게 정의할 수 있게 해줍니다.

에이전트 소프트웨어 (에이전트를 호출하는 소프트웨어)를 만드는 것은 복잡한 분산 시스템 (distributed systems) 문제입니다. ramure의 목표는 다음 두 가지 주목할 만한 방식으로 이러한 시스템을 더 쉽게 구축하고 견고하게 만드는 것입니다:

인프라 프리미티브 (Infrastructure primitives): 에이전트 통신, 프로비저닝 (provisioning), 그리고 에이전트가 실행되는 소프트웨어 환경을 위한 기능
결함 허용 및 모듈형 설계 (Fault-tolerant and modular design): ramure의 추상화 (abstractions) 및 이벤트 기반 로직 (event based logic)은 Erlang과 같은 분산 시스템 프로그래밍 (distributed systems programming)의 아이디어를 사용하여, 실패를 명확히 알리고 재시도하는 에이전트를 더 쉽게 작성할 수 있게 합니다. ramure 설계의 동기 부여를 이해하려면 멀티 에이전트 시스템 (multi-agent systems)을 분산 소프트웨어로 간주하십시오.

다음은 ramure가 에이전트를 통해 쉽게 수행할 수 있는 작업의 예시입니다:

• 최적화 (optimization)
• 사용자 입력이 포함된 맞춤형 소프트웨어 생성 파이프라인 (custom software generation pipelines)
• 데이터 파이프라인 (data pipelines)
• 워커 풀 (worker pools), 모니터 (monitors), 그리고 슈퍼바이저 (supervisors)

참고: ramure는 이전의 에이전트 런타임 (agent runtime)인 druids의 후속작입니다. 이전 코드는 fulcrumresearch/druids-archive에 보존되어 있습니다.

pip install ramure

Python 3.11+ 버전이 필요합니다.

ramure는 에이전트가 실행되는 머신을 위해 pi와 tmux에 의존합니다.

다음은 주석이 달린 단일 워커 (single worker) 프로그램 예시입니다:

import asyncio
from ramure import agent, agent_process, done, fail, wait
# 에이전트와 해당 머신을 관리하고 라이프사이클/정리 (lifecycle/cleanup)를 제어할 ramure 런타임을 등록합니다
...

다음 명령어로 실행하십시오:

uv run your_program.py

이제 ramure connect worker를 통해 에이전트에 연결하여 무엇을 하고 있는지 확인할 수 있습니다.

ramure의 핵심 객체는 agent_process (AP)이며, 비동기 함수 (async function)에 @agent_process 데코레이터 (decorator)를 붙여 정의합니다. 함수 내부에서 에이전트와 머신, 그리고 이들이 어떻게 통신해야 하는지를 정의합니다.

루트 AP (root AP)가 호출되면, ramure는 자신이 소유한 에이전트(agent)와 머신(machine)의 생명주기(lifecycle)를 관리하는 런타임(runtime)을 초기화합니다. 중첩된 AP들은 활성화된 런타임을 상속받습니다. AP의 생명주기를 제어하려면, @agent.on(...)을 통해 에이전트가 결정론적인 Python (deterministic Python)으로 콜백할 수 있는 이벤트를 정의합니다.

프로그램 내에서 정보가 이동하는 방식을 구조화하면, 특히 더 복잡한 사례에서 에이전트의 노동력을 안정적으로 사용하기가 더 쉬워집니다. 또한 AP가 실행될 image를 구성할 수도 있습니다. 기본값은 로컬 머신(local machine)이며, 다른 백엔드(backend)로 설정할 수도 있습니다 (Images and machines 참조).

AP는 합성(compose)됩니다. AP는 일반적인 비동기 함수(async function)를 호출하는 것과 같은 방식으로 다른 AP를 호출할 수 있습니다:

@agent_process
async def main():
    code = await write_code("fibonacci function")
    ...

또는 asyncio.gather를 사용하여 동시에 팬아웃(fan out)할 수도 있습니다:

@agent_process
async def main():
    results = await asyncio.gather(
        ...
    )

AP는 또한 spawn()을 호출하여 실행 중인 AP에 대한 핸들(handle)을 얻을 수 있으며, 해당 AP의 이벤트는 실시간으로 관찰 가능합니다.

@agent_process
async def main():
    handle = spawn(flaky_task, "write a haiku")
    ...

이 핸들은 자식 AP의 이벤트 스트림(event stream)을 유지하므로, 실행 중인 상태를 관찰하고, 실패 시 재시도(retry)하며, bubble()을 사용하여 상위 수준의 감독자(supervisor)에게 이벤트를 전달할 수 있습니다.

프로세스(Process)는 emit(type, data)를 통해 커스텀 이벤트를 방출(emit)할 수 있습니다. 만약 감독자가 자식의 이벤트를 자신의 이벤트 스트림에 나타나게 하고 싶다면 bubble()을 사용하십시오:

@agent_process
async def worker_pool(specs: list[str]) -> None:
    for i, spec in enumerate(specs):
        ...

이제 spawn(worker_pool, specs).events를 관찰하는 부모는 source=tid 태그가 붙은 자식 이벤트도 볼 수 있습니다.

에이전트는 실제로 어디에서 실행될까요? ramure에서 그것은 머신 (machine), 즉 실행 환경입니다. **이미지 (image)**는 이를 생성(spawn)하는 템플릿입니다.

Image.spawn() -> Machine

Machine은 네 가지 비동기 메서드(async methods)를 충족하는 모든 것을 의미합니다:

class Machine(ABC):
async def exec(self, command, *, user="agent", timeout=None) -> ExecResult: ...
async def write_file(self, path, content) -> None: ...
...

이것이 계약(contract)의 전부입니다. ramure는 머신(machine)의 tmux 세션에서 pi 하네스(harness)를 실행하기 위해 exec를 사용하고, 에이전트 확장(agent extension)을 그곳에 배치하기 위해 write_file을 사용하며, 정리를 위해 stop을 사용합니다. 두 가지 선택적 메서드인 fork()와 snapshot()은 백엔드(backends)가 저렴한 상태 복제(state duplication) 기능을 지원할 경우 이를 노출할 수 있게 해줍니다 (MorphCloud는 지원하지만, LocalMachine은 지원하지 않습니다).

번들로 제공되는 백엔드(Bundled backends):

LocalImage(workdir=, env=) — 사용자의 호스트(host)입니다. 기본값(Default)입니다. 생성된 각 LocalMachine은 단순히 작업 디렉토리(working directory)일 뿐입니다.
MorphImage(...) — MorphCloud VM입니다. 스냅샷(Snapshot)/포크(fork) 친화적입니다. pip install ramure[morph]로 설치할 수 있습니다.

새로운 백엔드(backend)는 한 쌍으로 구성됩니다: 머신을 띄우는 방법을 아는 Image와 실행 중인 대상을 감싸는(wraps) Machine입니다. 예를 들어, Docker 컨테이너 백엔드는 대략 다음과 같은 형태가 됩니다:

from ramure.machines.base import Image, Machine
from ramure.types import ExecResult
class DockerMachine(Machine):
...

ramure가 인스턴스를 받는 곳이라면 어디든 전달할 수 있습니다: @agent_process(image=DockerImage("my/image")), await agent("name", image=...), 또는 await machine(image=...).

런타임(runtime)은 머신 내부에서 WebSocket을 통해 에이전트를 등록하는 작업을 처리하며, 에이전트 프로세스(AP)가 종료되면 Machine.stop()을 통해 모든 것을 정리합니다.

가장 단순한 참조 구현(reference implementation)은 ramure/machines/local.py (~110줄)를, SSH, 스냅샷, 포크 기능이 포함된 구현은 ramure/machines/morph.py를 참고하십시오.

AP는 코드 내에서 호출하거나 다른 에이전트를 통해 호출할 수 있는 API를 노출함으로써, 상호작용하는 특정 방식을 인코딩할 수도 있습니다. 이를 위해 @expose 데코레이터(decorator)를 사용하십시오:

@agent_process
async def worker_pool() -> None:
specs: dict[str, str] = {}
...

그 후 다음과 같은 다양한 방식으로 노출된 워커 풀(worker pool)을 사용할 수 있습니다:

@agent_process
async def main():
pool = spawn(worker_pool)
...

이를 통해 컴포넌트에 모든 것에 대한 주변적 접근 권한 (ambient access)을 주는 대신, 좁은 허용 범위 (affordances)를 부여할 수 있습니다. 엔드포인트 (Endpoints)는 자식 프로세스의 스코프 (scope) 내에서 실행되므로, 엔드포인트 내부의 emit() , done() , fail() 호출은 호출자가 아닌 자식 프로세스에 영향을 미칩니다. 자식 프로세스가 소유한 에이전트 (agents) 또한 자식이 이를 생성한 후에는 handle.agents를 통해 확인할 수 있습니다.

@agent_process(image=, timeout=, log_dir=, host=, port=, base_url=)
— 비동기 함수 (async function)를 프로세스로 래핑 (wrap)

await agent(name, system_prompt=, model=, image=, machine=)
— 에이전트 생성 (model은 pi의 --model로 전달됨)

await machine(image=)
— 독립형 머신 (standalone machine) 생성

connect(a, b, direction=)
— 에이전트 간 메시지 전송/파일 전송 허용

done(result)
— 프로세스 성공 신호 전달

fail(reason)
— 프로세스 실패 신호 전달

await wait()
— done() 또는 fail()이 호출될 때까지 블로킹 (block)

emit(type, data)
— 프로세스 이벤트 방출 (emit)

spawn(fn, *args, **kwargs)
— 프로세스를 백그라운드에서 실행하며, ProcessHandle을 반환

bubble(handle, source=)
— 자식 프로세스의 이벤트를 현재 프로세스 스트림 (stream)으로 전달

@expose
— handle.call()을 통해 호출 가능하거나 handle.attach()를 통해 부착 가능한 엔드포인트로 비동기 함수를 등록

current_runtime()
— 활성 런타임 (active runtime)에 접근 (거의 필요하지 않음)

agent.on(tool_name)
— 비동기 툴 핸들러 (async tool handler)를 등록하기 위한 데코레이터 (decorator)

agent.send(message)
— 에이전트에게 메시지 전송

agent.exec(command)
— 에이전트의 머신에서 셸 명령 (shell command) 실행

agent.events
— 가공되지 않은 에이전트 이벤트의 비동기 반복 가능 (async-iterable) 로그

handle.events
— 프로세스 이벤트의 비동기 반복 가능 (async-iterable) 스트림

handle.agents
— 지금까지 생성된 자식의 에이전트들을 담은 딕셔너리 (dict)

await handle.call(name, **kwargs)
— 엔드포인트 호출

await handle.attach(agent, only=, prefix=)
— 엔드포인트를 에이전트의 툴 (tools)로 등록

handle.cancel()
— 프로세스 취소

루트 @agent_process를 실행하면 ~/.ramure/runtimes/{execution_id}.sock에 Unix 소켓을 열고, ~/.ramure/logs/{execution_id}/ 아래에 실행별 로그 트리 (log tree)를 작성합니다. ramure CLI는 이것들을 사용합니다:

ramure ls # 실행 중인 런 (live runs)
ramure status [--id <prefix>] # 에이전트 (agents), 머신 (machines), 연결 (connections), 어포던스 (affordances)
ramure send <agent> <msg> [--id <prefix>]
...

--id

실행 ID (execution-id) 접두사를 받습니다. 실행 중인 런 (live run)이 하나뿐이라면 생략할 수 있습니다. 모든 명령은 해당 런이 실행 중(소켓이 존재함)이어야 합니다. 종료된 런의 로그는 ~/.ramure/logs/{execution_id}/ 아래에 유지됩니다.

루트 @agent_process가 @expose 하는 모든 것은 ramure call을 통해 프로그램 외부에서 접근할 수 있습니다. 인자 (arguments)는 먼저 JSON으로 파싱되며, 실패 시 문자열 (string)로 처리되므로 플래그 없이도 타입이 지정된 값 (typed values)을 사용할 수 있습니다:

ramure call add_task spec="write tests"
ramure call add_task count=3 enabled=true tags='["a","b"]'
ramure call list_tasks --json

중첩된 AP @expose는 내부적으로 유지됩니다. 소유한 코드에서 handle.call 또는 handle.attach를 통해 접근하십시오. 오직 루트 프로그램의 표면 (surface)만이 외부에서 주소 지정이 가능합니다.