Cogency
상태가 없는 컨텍스트 조립 (stateless context assembly)을 통한 스트리밍 에이전트.
설치 (Install)
pip install cogency
export OPENAI_API_KEY="your-key"
빠른 시작 (Quickstart)
from cogency import Agent
agent = Agent(llm="openai")
...
핵심 설계 (Core Design)
- Persist-then-rebuild (저장 후 재구축): 이벤트가 즉시 스토리지에 기록되며, 매 실행 시 컨텍스트가 재구축됩니다.
- Protocol/storage separation (프로토콜/스토리지 분리): LLM I/O를 위한 XML 구분자(delimiters)를 사용하며, 스토리지에는 깔끔한 이벤트가 저장됩니다.
- Stateless execution (상태가 없는 실행): 에이전트와 컨텍스트는 순수 함수 (pure functions)이며, 모든 상태는 스토리지에 존재합니다.
결과: 상태 오염 없음, 충돌 복구 (crash recovery), 동시성 안전성 (concurrent safety).
실행 모드 (Execution Modes)
| 모드 | 방법 | 토큰 사용량 | 제공자 |
|---|
| Resume | WebSocket | 일정함 (Constant) | OpenAI, Gemini |
| ... | | | |
agent = Agent(llm="openai", mode="auto")
토큰 효율성 (Resume vs Replay):
| 턴 (Turns) | Replay | Resume | 절감액 |
|---|
| 16 | 100,800 | 10,800 | 9.3x |
| 32 | 355,200 | 20,400 | 17.4x |
스트리밍 (Streaming)
이벤트 모드 (Event mode, 기본값): 완전한 의미 단위 (Complete semantic units)
async for event in agent("Debug this code", stream="event"):
if event["type"] == "think":
print(f"~ {event['content']}")
...
토큰 모드 (Token mode): 실시간 스트리밍
async for event in agent("Debug this code", stream="token"):
if event["type"] == "respond":
print(event["content"], end="", flush=True)
대화 (Conversations)
상태가 없는 방식 (Stateless, 기본값):
async for event in agent("What's in this directory?"):
if event["type"] == "respond":
print(event["content"])
프로필 학습을 포함한 상태 유지 방식 (Stateful with profile learning):
async for event in agent(
"Continue our code review",
conversation_id="review_session",
...
| 도구 | 설명 |
|---|
read | 파일 읽기 (선택적 페이지네이션 포함) |
| ... | |
from dataclasses import dataclass
from typing import Annotated
from cogency import ToolResult
...
설정 (Configuration)
agent = Agent(
llm="openai",
mode="auto",
...
히스토리 압축 (History compression): 긴 대화의 경우, history_transform을 전달하여 컨텍스트를 압축할 수 있습니다:
async def compress(messages: list[dict]) -> list[dict]:
if len(messages) <= 20:
return messages
...
문서 (Documentation)
라이선스 (License)
Apache 2.0