moduagent 0.1.0

Composable Python runtime for building production AI agents

ModuAgent

PDF 기술 설계에 따라 구현한 합성형 Python AI Agent 프레임워크입니다. Agent/AgentRuntime, vLLM·Ollama, Function Tool, 토큰 스트리밍, 구조화 출력, RBAC, 대화·체크포인트 저장, Plan-and-Execute, 관측성 컴포넌트를 제공합니다.

설치

Python 3.10 이상이 필요합니다.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -e '.[test]'

Redis 저장소를 사용할 때는 비동기 Redis 클라이언트를 별도로 설치합니다.

python -m pip install redis

기본 사용

타입 힌트는 Tool 입력 스키마가 되고 docstring은 설명이 됩니다. idempotent=True인 Tool만 설정된 횟수만큼 재시도됩니다.

import asyncio

from moduagent import Agent, AgentConfig, RetryConfig, VLLMClient, function_tool


@function_tool(idempotent=True, timeout_seconds=5.0, max_result_bytes=4096)
def add(a: int, b: int) -> int:
    """두 정수를 더한다."""
    return a + b


async def main() -> None:
    model = VLLMClient(
        base_url="http://vllm.internal:8000/v1",
        model="company-model",
        api_key="internal-token",
    )
    agent = Agent(
        config=AgentConfig(
            name="calculator",
            instructions="필요하면 계산 도구를 사용해 간결하게 답한다.",
            retry=RetryConfig(max_attempts=2),
        ),
        model=model,
        tools=[add],
    )

    result = await agent.run("12와 30을 더해줘", session_id="demo-session")
    if result.error:
        raise RuntimeError(result.error)
    print(result.output)


asyncio.run(main())

Ollama는 모델 객체만 바꾸면 됩니다. VLLMClient, OllamaClient, Agent 생성자는 키워드 인자를 사용합니다.

from moduagent import OllamaClient

model = OllamaClient(
    base_url="http://localhost:11434",
    model="qwen3:14b",
)

토큰 스트리밍

Agent.stream()은 토큰뿐 아니라 모델·Tool·정책·종료 이벤트를 순서대로 전달합니다. 토큰은 MODEL_DELTA, 최종 AgentResult는 마지막 RUN_COMPLETED 또는 RUN_FAILED 이벤트에 들어 있습니다.

from moduagent import EventType

result = None
async for event in agent.stream("짧게 소개해줘", session_id="stream-session"):
    if event.type is EventType.MODEL_DELTA:
        print(event.data["delta"], end="", flush=True)
    elif event.type in (EventType.RUN_COMPLETED, EventType.RUN_FAILED):
        result = event.data["result"]

print()
if result is not None and result.error:
    raise RuntimeError(result.error)

Pydantic 구조화 출력

모델이 JSON Schema 구조화 출력을 지원해야 합니다. 성공 시 result.output은 문자열이 아니라 지정한 Pydantic 객체입니다.

from pydantic import BaseModel, Field

from moduagent import Agent, AgentConfig, PydanticOutputCodec


class Answer(BaseModel):
    answer: str
    confidence: float = Field(ge=0, le=1)


structured_agent = Agent(
    config=AgentConfig(
        name="structured-answer",
        instructions="반드시 요청된 JSON 스키마로 답한다.",
    ),
    model=model,
    output_codec=PydanticOutputCodec(model=Answer),
)

result = await structured_agent.run("한국의 수도는?")
print(result.output.answer, result.output.confidence)

RBAC Tool 권한

기본 Authorizer는 모든 Tool을 허용합니다. 운영 환경에서는 역할별 Tool allowlist를 명시하고, 실행 시 user_context.roles를 전달합니다. 등록되지 않은 역할과 Tool은 거부됩니다.

from moduagent import Agent, AgentConfig, RBACToolAuthorizer

authorizer = RBACToolAuthorizer(
    role_permissions={
        "calculator_user": {"add"},
        "admin": {"*"},
    }
)

secured_agent = Agent(
    config=AgentConfig(
        name="secured-calculator",
        instructions="허용된 계산 도구만 사용한다.",
    ),
    model=model,
    tools=[add],
    tool_authorizer=authorizer,
)

result = await secured_agent.run(
    "1과 2를 더해줘",
    session_id="rbac-session",
    user_context={"roles": ["calculator_user"]},
)

Redis 대화·체크포인트와 재개

Redis 클라이언트는 동기·비동기 방식 모두 주입할 수 있습니다. 아래 예시는 redis.asyncio를 사용합니다. 대화 TTL과 체크포인트 TTL은 서로 독립적입니다.

from redis.asyncio import Redis

from moduagent import (
    Agent,
    AgentConfig,
    RedisCheckpointStore,
    RedisConversationStore,
)

redis = Redis.from_url("redis://localhost:6379/0", decode_responses=True)
conversations = RedisConversationStore(
    client=redis,
    ttl_seconds=7 * 24 * 60 * 60,
)
checkpoints = RedisCheckpointStore(
    client=redis,
    ttl_seconds=24 * 60 * 60,
)

resumable_agent = Agent(
    config=AgentConfig(
        name="resumable-agent",
        instructions="요청을 끝까지 수행한다.",
    ),
    model=model,
    conversation_store=conversations,
    checkpoint_store=checkpoints,
)

failed = await resumable_agent.run("작업을 수행해줘", session_id="session-42")
if failed.error and await checkpoints.load(failed.run_id) is not None:
    resumed = await resumable_agent.resume(
        failed.run_id,
        session_id="session-42",
    )

재개 시에는 원래 실행과 같은 session_id, 호환되는 Agent 설정·Policy·Tool 구성을 사용해야 합니다. 완료된 실행의 체크포인트는 자동 삭제됩니다.

Plan-and-Execute

복수 단계 작업은 LLMPlanGenerator와 PlanAndExecutePolicy를 조합합니다. 계획은 policy_state에 저장되므로 체크포인트와 함께 복구됩니다.

from moduagent import (
    Agent,
    AgentConfig,
    LLMPlanGenerator,
    PlanAndExecutePolicy,
)

planning_agent = Agent(
    config=AgentConfig(
        name="research-agent",
        instructions="현재 계획 단계에 맞춰 실행하고 간결하게 답한다.",
    ),
    model=model,
    tools=[add],
    decision_policy=PlanAndExecutePolicy(
        plan_generator=LLMPlanGenerator(model=model, max_steps=4),
        revise_on_tool_failure=True,
    ),
)

관측성

실행 이벤트는 여러 Sink로 동시에 보낼 수 있습니다. 기본 로깅·감사 Sink는 비밀 키를 재귀적으로 마스킹하며, Sink 실패는 Agent 실행을 중단하지 않습니다.

from moduagent import (
    Agent,
    AgentConfig,
    AuditEventSink,
    CompositeEventSink,
    InMemoryMetricRecorder,
    LoggingEventSink,
    MetricsEventSink,
)

metrics = InMemoryMetricRecorder()
audit = AuditEventSink()  # 운영에서는 writer=...를 주입
events = CompositeEventSink(
    sinks=[
        LoggingEventSink(),
        MetricsEventSink(recorder=metrics),
        audit,
    ]
)

observed_agent = Agent(
    config=AgentConfig(name="observed-agent", instructions="간결하게 답한다."),
    model=model,
    event_sink=events,
)

LoggingEventSink, MetricsEventSink, AuditEventSink 외에도 EventSink.publish(event) 계약을 구현해 사내 로그·메트릭·감사 시스템을 연결할 수 있습니다.

테스트

네트워크 없이 전체 단위·런타임 통합 테스트를 실행합니다.

python3 -m pytest -q tests --ignore=tests/integration

실제 모델 서버 연동 테스트는 환경 변수를 설정한 뒤 선택 실행합니다.

VLLM_BASE_URL=http://localhost:8000/v1 \
VLLM_MODEL=company-model \
VLLM_API_KEY=token \
python3 -m pytest -q tests/integration/test_live_models.py -k vllm

OLLAMA_BASE_URL=http://localhost:11434 \
OLLAMA_MODEL=qwen3:14b \
python3 -m pytest -q tests/integration/test_live_models.py -k ollama

환경 변수가 없으면 해당 live 테스트는 skip됩니다.

SQLite repository 계약은 별도 서비스 없이 실행되며, Redis는 REDIS_URL이 설정된 경우에만 연결합니다.

python3 -m pytest -q tests/integration/test_live_persistence.py -k database

REDIS_URL=redis://localhost:6379/15 \
python3 -m pytest -q tests/integration/test_live_persistence.py -k redis

현재 보장 범위

• 체크포인트는 실행 상태 재개를 지원하지만 Tool 부작용의 exactly-once 실행을 보장하지 않습니다. 쓰기 Tool은 자체 idempotency key와 중복 처리 방어를 구현해야 합니다.
• idempotent=True는 프레임워크의 Tool 재시도를 허용한다는 의미이며 exactly-once 보장이 아닙니다.
• 동일 세션 직렬화는 한 AgentRuntime 프로세스 안에서만 적용됩니다. 분산 lock, 분산 실행 큐, worker scheduler는 현재 범위에 포함되지 않습니다.
• Redis 대화 저장은 list 명령을 지원하는 클라이언트에서 append 원자성을 사용하지만, 전체 Agent 실행의 분산 트랜잭션을 제공하지 않습니다.