ayushi-agarwall/graphagent

그래프 이론 (Graph theory)을 기반으로 구축된 의존성 없는 (Zero-dependency), 비동기 우선 (Async-first) 에이전트 오케스트레이션 (Orchestration) 프레임워크입니다.

GraphAgent는 AI 에이전트 및 복잡한 멀티 에이전트 오케스트레이션을 구축하기 위한 최소한의, 프로덕션 준비 완료된 (Production-ready) 프레임워크입니다. 핵심 코어 전체가 표준 라이브러리만을 사용하여 100줄 미만의 Python 코드로 작성되었습니다.

프레임워크 (Framework)	핵심 코드 라인 (Core Lines)	의존성 (Dependencies)	벤더 종속성 (Vendor Lock-in)
LangGraph	~15,000+	20+	LangChain 생태계
...

제로 블로트 (Zero bloat): 불필요한 추상화 없음
제로 의존성 (Zero dependencies): Python 표준 라이브러리만 사용
제로 벤더 종속성 (Zero vendor lock-in): 어떤 LLM, 도구 또는 서비스와도 통합 가능
네이티브 컨텍스트 그래프 (Native Context Graphs): 에이전트 상태 전이 (State transitions)가 탐색 가능한 그래프를 형성하여, 고급 디버깅 및 향후 GNN (Graph Neural Network) 기반 최적화를 위한 구조화된 데이터를 제공
진정한 병렬 실행 (True Parallel Execution): 노드의 동시 실행을 위해 asyncio.gather() 사용
표현식 캐싱 (Expression Caching): DSL (Domain Specific Language)을 한 번 파싱하여 컴파일된 실행 계획을 재사용
실행 트레이싱 (Execution Tracing) (출시 예정): 제한된 메모리 관리를 갖춘 타임스탬프 기반 트레이스 로그
입력 검증 (Input Validation): 실행 전 오류 포착 (누락된 노드, 잘못된 구문, 순환 흐름)
프로덕션 준비 완료 (Production Ready): 비동기 안전한 (Async-safe) 상태, 적절한 에러 핸들링, 타임아웃/재시도 지원

GraphAgent는 세 가지 기본 요소 (Primitives)를 기반으로 구축되었습니다:

구성 요소 (Component)	목적 (Purpose)
Node	작업의 원자적 단위 (에이전트)
State	타임스탬프 트레이싱을 포함한 스레드 안전한 (Thread-safe) 일시적 데이터 컨테이너
Flow	DSL 파서를 갖춘 그래프 오케스트레이터

>> 순차적 (Sequential): 조건 없이 A를 실행한 후 B를 실행, B의 결과 반환
& 병렬 (Parallel): A와 B를 동시에 실행 (asyncio.gather를 통한 진정한 병렬성)
? 조건부 성공 (Conditional Success): A가 성공할 경우에만 B를 실행 (False일 경우 단락 평가/short-circuits)
...

git clone https://github.com/ayushi-agarwall/tinyagent.git
cd tinyagent
pip install -e .

curl -O https://raw.githubusercontent.com/ayushi-agarwall/tinyagent/main/src/tinyagent/core.py

pip install tinyagent

import asyncio
from tinyagent import Node, State, Flow
# 노드를 직접 정의 (데코레이터 필요 없음)
...

TinyAgent의 그래프 기반 실행 모델(graph-based execution model)은 다음과 같은 기능을 기본적으로 지원합니다:

Fan-out 및 Fan-in을 활용한 병렬 실행 (Parallel execution): 여러 개의 특화된 에이전트(specialized agents)가 하위 작업(subtasks)을 동시에 수행합니다.

조건부 분기(conditional branching)를 포함한 순차적 파이프라인 (Sequential pipelines): 작업이 순서대로 실행되며, 성공/실패 여부에 따라 경로가 라우팅(routing)됩니다.

순차적 구성을 통한 검색 증강 생성 (Retrieval-augmented generation, RAG): 임베딩(embedding), 검색(retrieval), 재순위화(reranking), 생성(generation) 과정을 순차적으로 결합합니다.

병렬 추론 분기 (Parallel reasoning branches): 여러 가지 해결 경로를 동시에 탐색합니다.

라우터 노드 (Router nodes): 라우터 노드가 병렬로 실행되는 특화된 전문가 노드(expert nodes)로 작업을 전달하며, 결합기(combiner)가 최종 출력을 생성합니다.

<N>을 사용한 양방향 루프 (Bidirectional loops): 생성기(Generator)가 출력을 생성하면 검증기(validator)가 이를 확인하며, 성공하거나 최대 반복 횟수에 도달할 때까지 루프가 지속됩니다.

# 예시: 최대 3회 시도하는 자기 수정 (Self-correction)
await flow.run("generator <3> validator", state)

중첩된 플로우 (Nested flows): 오케스트레이터 노드(orchestrator nodes)가 하위 플로우(sub-flows)를 호출하여 트리 구조의 에이전트 계층 구조(agent hierarchies)를 형성할 수 있습니다.

TinyAgent는 에이전트 상태 전이(state transitions)를 탐색 가능한 그래프(traversable graph)로 취급하도록 설계되었습니다. 각 실행은 노드 이름, 상태(OK/TIMEOUT/ERR), 소요 시간(duration)을 포함하는 타임스탬프가 찍힌 트레이스 엔트리(trace entries)를 캡처하며, 이는 디버깅, 성능 분석 및 GNN(Graph Neural Network) 기반 최적화에 직접 사용할 수 있는 구조화된 인접 리스트(adjacency list)를 형성합니다.

계획된 API (Planned API):

# 출시 예정 — 아직 구현되지 않음
state = State(trace_id="workflow-123", max_trace=1000)
await flow.run("A >> (B & C) >> D", state)
...

State(
    data: dict[str, Any] | None = None, # 초기 상태 데이터
    async_safe: bool = False, # 동시 접근을 위한 비동기 잠금(async lock) 활성화
    ...
)

메서드 (Methods):

await state.get(key, default=None)

• 값 가져오기

await state.set(key, value)

• 값 저장하기

await state.update(key, fn)

• 원자적 읽기-수정-쓰기 (Atomic read-modify-write) (예: 카운터 증가)

출시 예정 (Coming soon):

State(trace_id=..., max_trace=...)
— 제한된 실행 트레이싱 (bounded execution tracing)

state.log(entry, metadata=None)
— 사용자 정의 트레이스 엔트리

state.trace
— (timestamp, event, metadata) 튜플의 데크(deque)

state.trace_id
— 고유 트레이스 식별자 (unique trace identifier)

Node(
name: str, # 고유 노드 식별자 (Unique node identifier)
fn: Callable[[State], Awaitable[bool]], # 비동기 함수 (Async function)
...

재시도 의미론 (Retry semantics):

시나리오 (Scenario)	retries (기본값)	retry_on_false=True
예외 발생 (Exception raised)	retries 횟수만큼 재시도	retries 횟수만큼 재시도
False 반환 (Returns False)	중단 후 False 반환	retries 횟수만큼 재시도
True 반환 (Returns True)	True 반환	True 반환
타임아웃 (Timeout)	retries 횟수만큼 재시도	retries 횟수만큼 재시도

결함 허용 (fault-tolerance) (일시적인 네트워크 오류 등)을 위해서는 retries만 사용하세요.
노드가 일시적으로 False를 반환할 수 있고 재시도가 필요한 경우에는 retry_on_false=True를 추가하세요.
두 노드 간의 조정된 왕복(back-and-forth) 동작이 필요한 경우에는 대신 <N> 루프 연산자(loop operator)를 사용하세요.

# 예외 발생 시에만 재시도 (기본값)
Node("fetch", fetch_fn, retries=3)

# False 반환 시에도 재시도 (예: 준비될 때까지 폴링)
...

Flow()

메서드 (Methods):

await flow.run(expr: str, state: State) -> bool

• DSL 표현식 실행

특징 (Features):

• 표현식 캐싱 (Expression caching): 한 번 파싱한 후 컴파일된 계획(compiled plan)을 재사용
• 검증 (Validation): 누락된 노드, 일치하지 않는 괄호 등을 확인
• 에러 핸들링 (Error handling): 사용 가능한 노드 목록을 포함한 명확한 에러 메시지 제공

<N> 연산자는 두 노드 사이의 실행을 교대로 수행합니다:

await flow.run("generator <3> reviewer", state)

실행 (Execution):

• generator 실행
• reviewer 실행
• 만약 reviewer가 True를 반환하면 종료 (성공) - 그렇지 않으면 최대 N번까지 반복
• 마지막 reviewer 결과 반환

사용 사례 (Use Case): generator가 출력을 생성하고 reviewer가 이를 검증하는 자기 수정 루프 (Self-correction loops).

& 연산자는 진정한 병렬성 (true parallelism)을 위해 asyncio.gather()를 사용합니다:

import time
async def api_call_1(state: State) -> bool:
    await asyncio.sleep(0.3)
    ...

? (조건부 성공 (Conditional Success)):

await flow.run("validate ? process", state)
# validate가 True를 반환할 경우에만 process를 실행

| (조건부 실패 (Conditional Failure)):

await flow.run("risky_task | fallback", state)
# risky_task가 False를 반환할 경우에만 fallback을 실행
# risky_task가 성공하면 risky_task의 결과를 반환

try:
    await flow.run("nonexistent_node", state)
except ValueError as e:
    ...

TinyAgent의 트레이스 로그(trace logs)는 Graph Neural Networks (GNN/GCN)의 학습 데이터로 사용하기 위해 특별히 설계된 타임스탬프 기반의 그래프 데이터로 구조화되어 있습니다. 이를 통해 다음과 같은 기능이 가능합니다:

실행 패턴 학습 (Execution Pattern Learning): GNN이 과거 트레이스로부터 최적의 실행 경로를 학습합니다.
시계열 모델링 (Temporal Modeling): 타임스탬프를 통해 시계열 그래프(temporal graphs) 상의 상태 공간 모델 (State Space Models)을 구현할 수 있습니다.
예측 라우팅 (Predictive Routing): 노드 성능에 기반한 동적 엣지 가중치 (edge weight) 최적화를 수행합니다.
이상 탐지 (Anomaly Detection): 최적화되지 않았거나 실패하는 흐름 패턴을 식별합니다.
자가 치유 흐름 (Self-Healing Flows): 예측된 장애 지점을 피해 자동으로 경로를 재설정합니다.

네이티브 컨텍스트 그래프 구조(인접 리스트로서의 트레이스)는 GNN 입력 형식에 직접적으로 매핑됩니다.

• 워크플로우 시각화 (트레이스로부터 생성된 Mermaid 다이어그램)
• OpenTelemetry 통합
• Pub/sub 이벤트 시스템

MIT License. 자세한 내용은 LICENSE를 참조하십시오.