OpenManus 코드 분석으로 알아보는 AI 에이전트 아키텍처😊

📝 OpenManus란?

OpenManus는 "초대 코드 없이도 모든 아이디어를 실현할 수 있는" AI 에이전트 프레임워크입니다. MetaGPT 기여자들이 개발한 이 프로젝트는 AI 에이전트를 쉽게 구축하고 활용할 수 있게 해주는 오픈소스 솔루션입니다.

🔍 코드 구조 분석

GitHub에서 프로젝트를 클론하고 코드를 살펴보면서 OpenManus의 구조를 파악해봤습니다.

1) 메인 진입점: main.py

import asyncio

from app.agent.manus import Manus
from app.logger import logger

async def main():
    agent = Manus()
    try:
        prompt = input("Enter your prompt: ")
        if not prompt.strip():
            logger.warning("Empty prompt provided.")
            return

        logger.warning("Processing your request...")
        await agent.run(prompt)
        logger.info("Request processing completed.")
    except KeyboardInterrupt:
        logger.warning("Operation interrupted.")

if __name__ == "__main__":
    asyncio.run(main())

이 코드는 매우 단순하지만 OpenManus의 기본 작동 방식을 보여줍니다:

1. Manus 에이전트 인스턴스를 생성합니다.
2. 사용자로부터 프롬프트를 입력받습니다.
3. 에이전트의 run 메서드를 호출하여 프롬프트를 처리합니다.
4. 비동기(async/await) 패턴을 사용하여 효율적인 처리를 지원합니다.

2) app/agent/init.py

from app.agent.base import BaseAgent
from app.agent.planning import PlanningAgent
from app.agent.react import ReActAgent
from app.agent.swe import SWEAgent
from app.agent.toolcall import ToolCallAgent

__all__ = [
    "BaseAgent",
    "PlanningAgent",
    "ReActAgent",
    "SWEAgent",
    "ToolCallAgent",
]

이 코드에서 OpenManus가 다양한 유형의 에이전트를 제공하는 것을 알 수 있습니다:

• BaseAgent: 모든 에이전트의 기본 클래스
• PlanningAgent: 계획 수립 기능에 특화된 에이전트
• ReActAgent: Reasoning and Acting 패턴 구현 (생각하고 행동하는 패턴)
• SWEAgent: 소프트웨어 엔지니어링 작업을 위한 에이전트
• ToolCallAgent: 외부 도구 호출에 중점을 둔 에이전트

3) app/agent/base.py

BaseAgent 클래스는 OpenManus 프레임워크에서 모든 에이전트의 기본 골격을 제공합니다. 상태 관리, 메모리 처리, 실행 루프 등 에이전트의 핵심 동작을 정의하고 있습니다.

from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
from typing import List, Optional

from pydantic import BaseModel, Field, model_validator

from app.llm import LLM
from app.logger import logger
from app.schema import ROLE_TYPE, AgentState, Memory, Message


class BaseAgent(BaseModel, ABC):
    """Abstract base class for managing agent state and execution.

    Provides foundational functionality for state transitions, memory management,
    and a step-based execution loop. Subclasses must implement the `step` method.
    """

    # Core attributes
    name: str = Field(..., description="Unique name of the agent")
    description: Optional[str] = Field(None, description="Optional agent description")

    # Prompts
    system_prompt: Optional[str] = Field(
        None, description="System-level instruction prompt"
    )
    next_step_prompt: Optional[str] = Field(
        None, description="Prompt for determining next action"
    )

    # Dependencies
    llm: LLM = Field(default_factory=LLM, description="Language model instance")
    memory: Memory = Field(default_factory=Memory, description="Agent's memory store")
    state: AgentState = Field(
        default=AgentState.IDLE, description="Current agent state"
    )

    # Execution control
    max_steps: int = Field(default=10, description="Maximum steps before termination")
    current_step: int = Field(default=0, description="Current step in execution")

    duplicate_threshold: int = 2

    class Config:
        arbitrary_types_allowed = True
        extra = "allow"  # Allow extra fields for flexibility in subclasses

주요 기능별 분석

1. 상속 및 구조

• BaseModel (Pydantic): 데이터 유효성 검증과 타입 안전성 제공
• ABC (Abstract Base Class): 추상 메서드 정의를 통한 인터페이스 보장

이 두 가지를 상속받음으로써 견고한 데이터 모델과 인터페이스를 동시에 보장합니다.

2. 필수 속성

• Core attributes: 에이전트의 기본 정보 (이름, 설명)
• Prompts: 에이전트 동작을 위한 프롬프트 (시스템 프롬프트, 다음 단계 프롬프트)
• Dependencies: 작동에 필요한 의존성 (LLM, 메모리, 상태)
• Execution control: 실행 제어를 위한 매개변수 (최대 단계, 현재 단계)

3. 초기화 및 유효성 검증

@model_validator(mode="after")
def initialize_agent(self) -> "BaseAgent":
    """Initialize agent with default settings if not provided."""
    if self.llm is None or not isinstance(self.llm, LLM):
        self.llm = LLM(config_name=self.name.lower())
    if not isinstance(self.memory, Memory):
        self.memory = Memory()
    return self

Pydantic의 model_validator를 사용하여 객체 생성 후 유효성을 검사하고 적절한 기본값을 설정합니다. 특히 LLM과 메모리 인스턴스를 자동으로 생성해주는 부분이 중요합니다.

4. 상태 관리

@asynccontextmanager
async def state_context(self, new_state: AgentState):
    """Context manager for safe agent state transitions."""
    if not isinstance(new_state, AgentState):
        raise ValueError(f"Invalid state: {new_state}")

    previous_state = self.state
    self.state = new_state
    try:
        yield
    except Exception as e:
        self.state = AgentState.ERROR  # Transition to ERROR on failure
        raise e
    finally:
        self.state = previous_state  # Revert to previous state

비동기 컨텍스트 매니저를 사용한 우아한 상태 전환 관리가 인상적입니다:

• 임시로 상태를 변경하고 작업 완료 후 원래 상태로 복원
• 예외 발생 시 ERROR 상태로 전환
• with 구문과 유사한 간결한 사용 패턴 제공

5. 메모리 관리

def update_memory(
    self,
    role: ROLE_TYPE,  # type: ignore
    content: str,
    **kwargs,
) -> None:
    """Add a message to the agent's memory."""
    message_map = {
        "user": Message.user_message,
        "system": Message.system_message,
        "assistant": Message.assistant_message,
        "tool": lambda content, **kw: Message.tool_message(content, **kw),
    }

    if role not in message_map:
        raise ValueError(f"Unsupported message role: {role}")

    msg_factory = message_map[role]
    msg = msg_factory(content, **kwargs) if role == "tool" else msg_factory(content)
    self.memory.add_message(msg)

팩토리 패턴을 사용한 메시지 생성 및 저장:

• 다양한 역할(user, system, assistant, tool)의 메시지 지원
• 메시지 생성과 메모리 저장을 캡슐화
• 도구 호출 메시지에 대한 특별 처리

6. 핵심 실행 루프

async def run(self, request: Optional[str] = None) -> str:
    """Execute the agent's main loop asynchronously."""
    if self.state != AgentState.IDLE:
        raise RuntimeError(f"Cannot run agent from state: {self.state}")

    if request:
        self.update_memory("user", request)

    results: List[str] = []
    async with self.state_context(AgentState.RUNNING):
        while (
            self.current_step < self.max_steps and self.state != AgentState.FINISHED
        ):
            self.current_step += 1
            logger.info(f"Executing step {self.current_step}/{self.max_steps}")
            step_result = await self.step()

            # Check for stuck state
            if self.is_stuck():
                self.handle_stuck_state()

            results.append(f"Step {self.current_step}: {step_result}")

        if self.current_step >= self.max_steps:
            self.current_step = 0
            self.state = AgentState.IDLE
            results.append(f"Terminated: Reached max steps ({self.max_steps})")

    return "\n".join(results) if results else "No steps executed"

비동기 실행 루프의 주요 특징:

• 상태 확인 (IDLE 상태에서만 실행 가능)
• 사용자 요청을 메모리에 저장
• RUNNING 상태로 전환하여 작업 수행
• 최대 단계까지 또는 FINISHED 상태가 될 때까지 반복 실행
• 각 단계에서 중복 응답(루프) 감지 및 처리
• 실행 결과를 문자열로 반환

7. 추상 메서드

@abstractmethod
async def step(self) -> str:
    """Execute a single step in the agent's workflow.
    
    Must be implemented by subclasses to define specific behavior.
    """

모든 하위 클래스가 구현해야 하는 핵심 메서드입니다. 이 메서드는 에이전트의 실제 동작 로직을 정의합니다.

8. 루프 감지 및 처리

def handle_stuck_state(self):
    """Handle stuck state by adding a prompt to change strategy"""
    stuck_prompt = "\
    Observed duplicate responses. Consider new strategies and avoid repeating ineffective paths already attempted."
    self.next_step_prompt = f"{stuck_prompt}\n{self.next_step_prompt}"
    logger.warning(f"Agent detected stuck state. Added prompt: {stuck_prompt}")

def is_stuck(self) -> bool:
    """Check if the agent is stuck in a loop by detecting duplicate content"""
    if len(self.memory.messages) < 2:
        return False

    last_message = self.memory.messages[-1]
    if not last_message.content:
        return False

    # Count identical content occurrences
    duplicate_count = sum(
        1
        for msg in reversed(self.memory.messages[:-1])
        if msg.role == "assistant" and msg.content == last_message.content
    )

    return duplicate_count >= self.duplicate_threshold

에이전트가 동일한 응답을 반복하는 경우를 감지하고 처리하는 기능:

• 이전 메시지와 동일한 내용이 반복되는지 확인
• 지정된 임계값 이상으로 중복되면 루프 상태로 판단
• 새로운 전략을 고려하도록 프롬프트 수정

9. 속성 관리

@property
def messages(self) -> List[Message]:
    """Retrieve a list of messages from the agent's memory."""
    return self.memory.messages

@messages.setter
def messages(self, value: List[Message]):
    """Set the list of messages in the agent's memory."""
    self.memory.messages = value

메모리 내 메시지 접근을 위한 속성 래퍼:

• 캡슐화 원칙을 지키면서 메시지 접근 제공
• 속성처럼 사용할 수 있는 인터페이스 제공

설계 패턴 및 원칙

BaseAgent 클래스에서 볼 수 있는 주요 설계 패턴과 원칙:

1. 템플릿 메서드 패턴: run 메서드가 실행 흐름을 정의하고, 하위 클래스가 step 메서드를 구현
2. 팩토리 패턴: 메시지 생성에 팩토리 메서드 사용
3. 컨텍스트 매니저 패턴: 상태 전환 관리에 비동기 컨텍스트 매니저 사용
4. 캡슐화: 내부 상태와 메모리 접근 메서드 제공
5. 유효성 검증: Pydantic을 활용한 데이터 유효성 검사
6. 인터페이스 분리: 추상 기본 클래스를 통한 인터페이스 정의

4) app/agent/manus.py

Manus 클래스 분석

Manus 클래스는 ToolCallAgent를 상속받고 있으며, 이는 앞서 본 구조에서 BaseAgent의 하위 클래스입니다. 이 클래스는 다양한 도구들을 활용하여 범용적인 작업을 처리할 수 있는 AI 에이전트를 구현합니다.

주요 특징

1. 상속 관계:

   class Manus(ToolCallAgent):

   ToolCallAgent를 상속받아 도구 호출 기능을 확장합니다.

2. 기본 정보:

   name: str = "Manus"
   description: str = "A versatile agent that can solve various tasks using multiple tools"

   에이전트의 이름과 설명을 정의합니다.

3. 프롬프트 설정:

   system_prompt: str = SYSTEM_PROMPT
   next_step_prompt: str = NEXT_STEP_PROMPT

   app.prompt.manus 모듈에서 가져온 미리 정의된 프롬프트를 사용합니다.

4. 실행 제한:

   max_observe: int = 2000
   max_steps: int = 20

   관찰 데이터 길이와 최대 실행 단계 수를 제한합니다.

5. 도구 모음:

   available_tools: ToolCollection = Field(
       default_factory=lambda: ToolCollection(
           PythonExecute(), WebSearch(), BrowserUseTool(), FileSaver(), Terminate()
       )
   )

   에이전트가 사용할 수 있는 도구 모음을 정의합니다:

   • PythonExecute: Python 코드 실행
   • WebSearch: 웹 검색
   • BrowserUseTool: 웹 브라우저 조작
   • FileSaver: 파일 저장
   • Terminate: 작업 종료

6. 특별 도구 처리:

   async def _handle_special_tool(self, name: str, result: Any, **kwargs):
       if not self._is_special_tool(name):
           return
       else:
           await self.available_tools.get_tool(BrowserUseTool().name).cleanup()
           await super()._handle_special_tool(name, result, **kwargs)

   특별 도구를 처리하는 메서드를 오버라이드하여, 특별 도구 실행 전에 브라우저 도구의 클린업을 수행합니다.

분석 및 의의

1. 다양한 기능을 갖춘 범용 에이전트:
   Manus 클래스는 코드 실행, 웹 검색, 브라우저 조작, 파일 저장 등 다양한 도구를 결합한 범용 에이전트입니다.

2. 도구 추상화:
   ToolCollection을 통해 여러 도구를 관리하고, 각 도구는 독립적인 클래스로 구현되어 확장성이 뛰어납니다.

3. 특별 도구 처리 메커니즘:
   특정 도구들에 대한 특별 처리 로직을 구현하여 자원 관리(브라우저 정리 등)를 자동화합니다.

4. 상위 클래스 활용:
   super()._handle_special_tool()를 호출하여 상위 클래스의 기능을 재사용하는 좋은 OOP 관행을 보여줍니다.

5) app/agent/planning.py

PlanningAgent 클래스 개요

PlanningAgent는 작업을 해결하기 위한 구조화된 계획을 생성하고 관리하는 에이전트입니다. 계획의 각 단계를 추적하고 완료할 때까지 진행합니다.

주요 특징 및 기능

1. 기본 속성 설정:

   • 이름, 설명, 프롬프트 등 기본 정보 정의
   • PlanningTool과 Terminate 도구를 기본적으로 사용

2. 계획 관리 속성:

   • active_plan_id: 현재 활성화된 계획의 ID
   • step_execution_tracker: 도구 호출별로 단계 실행 상태를 추적하는 딕셔너리
   • current_step_index: 현재 실행 중인 단계의 인덱스

3. 초기화 및 유효성 검증:

   @model_validator(mode="after")
   def initialize_plan_and_verify_tools(self) -> "PlanningAgent":
       self.active_plan_id = f"plan_{int(time.time())}"
       if "planning" not in self.available_tools.tool_map:
           self.available_tools.add_tool(PlanningTool())
       return self

   • 시간 기반 고유 계획 ID 생성
   • 필요한 도구가 없는 경우 자동 추가

4. 사고 과정 (think 메서드):

   async def think(self) -> bool:
       # 현재 계획 상태를 기반으로 다음 행동 결정
       prompt = f"CURRENT PLAN STATUS:\n{await self.get_plan()}\n\n{self.next_step_prompt}"
       # ...

   • 현재 계획 상태를 포함한 프롬프트 생성
   • 현재 단계 인덱스 파악
   • 도구 호출과 현재 단계를 연결하여 추적

5. 행동 실행 (act 메서드):

   async def act(self) -> str:
       # 단계 실행 및 완료 상태 추적
       # ...

   • 도구 실행 후 단계 완료 상태 업데이트
   • 비계획, 비특수 도구인 경우 계획 상태 업데이트

6. 계획 관리 메서드:

   • get_plan(): 현재 계획 상태 조회
   • update_plan_status(): 도구 실행 후 계획 상태 업데이트
   • _get_current_step_index(): 첫 번째 미완료 단계 인덱스 찾기
   • create_initial_plan(): 초기 요청을 기반으로 계획 생성

7. 실행 메서드 (run):

   async def run(self, request: Optional[str] = None) -> str:
       if request:
           await self.create_initial_plan(request)
       return await super().run()

   • 요청이 있으면 초기 계획 생성
   • 상위 클래스의 run 메서드 호출

작동 방식

1. 사용자 요청을 받으면 create_initial_plan을 통해 계획 생성
2. run 메서드에서 계획 실행 시작
3. think 단계에서 현재 계획 상태를 기반으로 다음 행동 결정
4. act 단계에서 행동 실행 및 결과 처리
5. 도구 실행 후 계획 상태 업데이트
6. 모든 단계가 완료되거나 최대 단계에 도달할 때까지 반복

핵심 설계 특징

1. 상태 추적: 각 도구 호출과 계획 단계를 연결하여 진행 상황 추적
2. 단계별 진행: 계획의 각 단계를 순차적으로 처리
3. 계획 시각화: 현재 계획 상태를 사용자에게 보여주는 기능
4. 상태 업데이트: 도구 실행 후 계획의 단계 상태 자동 업데이트

이 코드는 AI 에이전트가 복잡한 작업을 여러 단계로 나누어 체계적으로 해결하는 방식을 잘 보여줍니다. 특히 계획 생성, 추적, 업데이트를 자동화하여 구조화된 문제 해결 접근법을 구현한 점이 인상적입니다.

6) app/agent/react.py

ReActAgent 클래스 분석

ReActAgent는 BaseAgent를 상속하고 추가적인 추상 메서드를 정의하는 추상 클래스(ABC)입니다.

주요 속성

name: str                                # 에이전트 이름
description: Optional[str] = None        # 에이전트 설명
system_prompt: Optional[str] = None      # 시스템 프롬프트
next_step_prompt: Optional[str] = None   # 다음 단계 프롬프트
llm: Optional[LLM] = Field(default_factory=LLM)    # 언어 모델 인스턴스
memory: Memory = Field(default_factory=Memory)     # 메모리 객체
state: AgentState = AgentState.IDLE      # 현재 상태
max_steps: int = 10                      # 최대 단계 수
current_step: int = 0                    # 현재 단계

이 속성들은 앞서 본 BaseAgent의 속성과 유사하며, ReActAgent는 특히 ReAct 패턴을 구현하기 위한 기반을 제공합니다.

주요 메서드

1. 추상 메서드 think

   @abstractmethod
   async def think(self) -> bool:
       """Process current state and decide next action"""

   • 현재 상태를 처리하고 다음 행동을 결정하는 메서드
   • 반환값은 행동을 수행해야 하는지 여부를 나타냄 (True/False)
   • 하위 클래스에서 구현해야 함

2. 추상 메서드 act

   @abstractmethod
   async def act(self) -> str:
       """Execute decided actions"""

   • 결정된 행동을 실행하는 메서드
   • 하위 클래스에서 구현해야 함

3. step 메서드

   async def step(self) -> str:
       """Execute a single step: think and act."""
       should_act = await self.think()
       if not should_act:
           return "Thinking complete - no action needed"
       return await self.act()

   • BaseAgent의 추상 메서드 step을 구현한 메서드
   • think 메서드를 호출하여 행동 여부 결정
   • think가 True를 반환하면 act 메서드를 호출하여 행동 수행
   • 결과를 문자열로 반환

ReAct 패턴의 구현

이 클래스는 ReAct(Reasoning + Acting) 패턴을 구현한 좋은 예입니다:

1. Reasoning 단계: think 메서드는 현재 상황을 분석하고 행동 계획을 세우는 추론 과정을 담당
2. Acting 단계: act 메서드는 계획에 따라 실제 행동을 수행
3. 반복적 실행: step 메서드를 통해 이 두 단계를 반복적으로 수행

설계의 특징

1. 추상화와 템플릿 메서드 패턴:

   • think와 act는 추상 메서드로, 구체적인 구현은 하위 클래스에 위임
   • step은 템플릿 메서드로, 전체 흐름은 정의하되 세부 구현은 하위 클래스에서 결정

2. 유연성:

   • 다양한 종류의 추론 및 행동 방식을 구현할 수 있는 유연한 구조

3. 단순성:

   • 복잡한 AI 에이전트의 핵심 동작을 "think"와 "act"라는 두 가지 명확한 단계로 단순화

4. 비동기 프로그래밍:

   • 모든 메서드가 async로 선언되어 비동기 작업(예: API 호출)을 효율적으로 처리

이 ReActAgent 클래스는 다양한 목적의 구체적인 에이전트를 구현하기 위한 기반을 제공하며, BaseAgent에서 정의한 에이전트의 기본 골격에 ReAct 패턴을 접목시킨 구조를 보여줍니다.

7) app/agent/swe.py

SWEAgent 클래스 분석

SWEAgent는 ToolCallAgent를 상속받은 클래스로, 프로그래밍 및 소프트웨어 엔지니어링 작업을 수행하기 위한 도구들을 갖추고 있습니다.

주요 속성

name: str = "swe"
description: str = "an autonomous AI programmer that interacts directly with the computer to solve tasks."

system_prompt: str = SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_TEMPLATE

available_tools: ToolCollection = ToolCollection(
    Bash(), StrReplaceEditor(), Terminate()
)
special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name])

max_steps: int = 30

bash: Bash = Field(default_factory=Bash)
working_dir: str = "."

1. 기본 정보:

   • 이름: "swe"
   • 설명: 컴퓨터와 직접 상호작용하여 작업을 해결하는 자율 AI 프로그래머

2. 프롬프트:

   • SYSTEM_PROMPT: 소프트웨어 엔지니어링 작업에 특화된 시스템 프롬프트
   • NEXT_STEP_TEMPLATE: 작업 디렉토리 정보를 포함할 수 있는 템플릿 형태의 프롬프트

3. 도구 모음:

   • Bash(): 셸 명령어 실행 도구
   • StrReplaceEditor(): 텍스트/코드 수정 도구
   • Terminate(): 작업 종료 도구

4. 실행 제어:

   • max_steps: 30 (다른 에이전트보다 많은 단계 허용)
   • bash: Bash 도구의 인스턴스
   • working_dir: 현재 작업 디렉토리 (기본값: ".")

주요 메서드

async def think(self) -> bool:
    """Process current state and decide next action"""
    # Update working directory
    self.working_dir = await self.bash.execute("pwd")
    self.next_step_prompt = self.next_step_prompt.format(
        current_dir=self.working_dir
    )

    return await super().think()

think 메서드는 부모 클래스의 메서드를 오버라이드하여 다음과 같은 기능을 추가합니다:

1. pwd 명령어를 실행하여 현재 작업 디렉토리를 가져옴
2. 템플릿에 현재 디렉토리 정보를 삽입하여 next_step_prompt 업데이트
3. 상위 클래스의 think 메서드 호출

SWEAgent의 특징

1. 개발 환경 인식:

   • 현재 작업 디렉토리를 추적하고 프롬프트에 포함하여 에이전트가 파일 시스템 컨텍스트를 인식하도록 함

2. 코드 조작 도구:

   • Bash 명령어 실행 (파일 조작, 컴파일, 실행 등)
   • 문자열 교체 기반 에디터 (코드 수정)

3. 소프트웨어 개발 맥락:

   • 프로그래밍 및 소프트웨어 개발 작업에 특화된 프롬프트와 도구 구성

4. 긴 작업 허용:

   • max_steps가 30으로 설정되어 있어 복잡한 프로그래밍 작업을 수행할 수 있는 충분한 단계 제공

이 SWEAgent 클래스는 코드 작성, 디버깅, 소프트웨어 개발과 같은 프로그래밍 관련 작업을 수행하는 데 특화된 에이전트입니다. Bash 명령어와 텍스트 편집 도구를 통해 파일 시스템과 코드를 직접 조작할 수 있는 능력을 가지고 있으며, 이를 통해 복잡한 개발 작업을 자동화할 수 있습니다.

8) app/agent/toolcall.py

ToolCallAgent 클래스 분석

ToolCallAgent는 도구/함수 호출을 처리하기 위한 향상된 추상화를 제공하는 기본 에이전트 클래스입니다.

주요 속성

name: str = "toolcall"
description: str = "an agent that can execute tool calls."

system_prompt: str = SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT

available_tools: ToolCollection = ToolCollection(
    CreateChatCompletion(), Terminate()
)
tool_choices: TOOL_CHOICE_TYPE = ToolChoice.AUTO
special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name])

tool_calls: List[ToolCall] = Field(default_factory=list)

max_steps: int = 30
max_observe: Optional[Union[int, bool]] = None
max_input_tokens: Optional[int] = None

1. 기본 도구 설정:

   • CreateChatCompletion(): 대화 완성을 위한 도구
   • Terminate(): 작업 종료 도구

2. 도구 선택 모드:

   • tool_choices: 도구 선택 모드 (AUTO, REQUIRED, NONE)

3. 특수 도구 처리:

   • special_tool_names: 특별히 처리되는 도구 이름 목록

4. 도구 호출 저장:

   • tool_calls: 현재 처리 중인 도구 호출 목록

5. 제한 설정:

   • max_observe: 도구 출력 결과 길이 제한
   • max_input_tokens: 최대 입력 토큰 수 제한

주요 메서드

1. think 메서드

async def think(self) -> bool:
    """Process current state and decide next actions using tools"""
    # 다음 단계 프롬프트 추가
    if self.next_step_prompt:
        user_msg = Message.user_message(self.next_step_prompt)
        self.messages += [user_msg]

    try:
        # 도구 옵션으로 응답 얻기
        response = await self.llm.ask_tool(
            messages=self.messages,
            system_msgs=[Message.system_message(self.system_prompt)]
            if self.system_prompt
            else None,
            tools=self.available_tools.to_params(),
            tool_choice=self.tool_choices,
        )
    except Exception as e:
        # 토큰 제한 오류 처리
        # ...

    self.tool_calls = response.tool_calls

    # 다양한 도구 선택 모드 처리
    # ...

이 메서드는 다음과 같은 역할을 합니다:

• 다음 단계 프롬프트를 메모리에 추가
• LLM에게 도구 옵션으로 응답을 요청
• 토큰 제한 오류와 같은 예외 처리
• 도구 호출 결과 로깅
• 도구 선택 모드에 따른 응답 처리

2. act 메서드

async def act(self) -> str:
    """Execute tool calls and handle their results"""
    if not self.tool_calls:
        if self.tool_choices == ToolChoice.REQUIRED:
            raise ValueError(TOOL_CALL_REQUIRED)

        # 도구 호출이 없으면 마지막 메시지 내용 반환
        return self.messages[-1].content or "No content or commands to execute"

    results = []
    for command in self.tool_calls:
        result = await self.execute_tool(command)

        if self.max_observe:
            result = result[: self.max_observe]

        # 도구 응답을 메모리에 추가
        tool_msg = Message.tool_message(
            content=result, tool_call_id=command.id, name=command.function.name
        )
        self.memory.add_message(tool_msg)
        results.append(result)

    return "\n\n".join(results)

이 메서드는 다음과 같은 역할을 합니다:

• 도구 호출이 없는 경우 처리
• 각 도구 명령 실행
• 결과 길이 제한 적용
• 도구 응답을 메모리에 추가
• 모든 결과 병합

3. execute_tool 메서드

async def execute_tool(self, command: ToolCall) -> str:
    """Execute a single tool call with robust error handling"""
    if not command or not command.function or not command.function.name:
        return "Error: Invalid command format"

    name = command.function.name
    if name not in self.available_tools.tool_map:
        return f"Error: Unknown tool '{name}'"

    try:
        # 인자 파싱
        args = json.loads(command.function.arguments or "{}")

        # 도구 실행
        result = await self.available_tools.execute(name=name, tool_input=args)

        # 표시용 결과 형식 지정
        observation = f"Observed output of cmd `{name}` executed:\n{str(result)}"

        # `finish`와 같은 특수 도구 처리
        await self._handle_special_tool(name=name, result=result)

        return observation
    except json.JSONDecodeError:
        # JSON 파싱 오류 처리
    except Exception as e:
        # 일반 오류 처리

이 메서드는 다음과 같은 역할을 합니다:

• 명령 유효성 검사
• JSON 인자 파싱
• 도구 실행
• 결과 형식 지정
• 특수 도구 처리
• 오류 처리

4. 특수 도구 처리 메서드

async def _handle_special_tool(self, name: str, result: Any, **kwargs):
    """Handle special tool execution and state changes"""
    if not self._is_special_tool(name):
        return

    if self._should_finish_execution(name=name, result=result, **kwargs):
        # 에이전트 상태를 완료로 설정
        self.state = AgentState.FINISHED

@staticmethod
def _should_finish_execution(**kwargs) -> bool:
    """Determine if tool execution should finish the agent"""
    return True

def _is_special_tool(self, name: str) -> bool:
    """Check if tool name is in special tools list"""
    return name.lower() in [n.lower() for n in self.special_tool_names]

이 메서드들은 특수 도구(예: Terminate)를 처리하는 방법을 정의합니다:

• 특수 도구 여부 확인
• 실행 종료 여부 결정
• 에이전트 상태 변경

설계 특징 및 패턴

1. 도구 기반 추론: LLM이 추론과 도구 선택을 모두 결정하게 함
2. 오류 처리: 다양한 예외 상황에 대한 견고한 처리
3. 모듈화: 각 단계(생각, 행동, 도구 실행)가 명확히 분리됨
4. 유연성: 다양한 도구 선택 모드(AUTO, REQUIRED, NONE) 지원
5. 상태 관리: 특수 도구 실행 시 에이전트 상태 변경
6. 결과 제한: 도구 출력 길이 제한을 통한 토큰 관리

이 ToolCallAgent 클래스는 OpenManus의 핵심 구성 요소로, LLM이 다양한 도구를 활용하여 복잡한 작업을 수행할 수 있게 하는 기반을 제공합니다. 이전에 본 SWEAgent, PlanningAgent, Manus 클래스들은 모두 이 ToolCallAgent를 상속받아 특정 용도에 맞게 확장한 것입니다.

9) app/flow/base.py

BaseFlow 클래스 분석

BaseFlow는 여러 에이전트를 지원하는 실행 흐름의 기본 클래스입니다.

주요 속성

agents: Dict[str, BaseAgent]
tools: Optional[List] = None
primary_agent_key: Optional[str] = None

1. 에이전트 관리:

   • agents: 이름을 키로 하는 에이전트 딕셔너리
   • primary_agent_key: 기본 에이전트를 식별하는 키

2. 도구 설정:

   • tools: 선택적으로 사용 가능한 도구 목록

주요 메서드

1. 초기화 메서드

def __init__(
    self, agents: Union[BaseAgent, List[BaseAgent], Dict[str, BaseAgent]], **data
):
    # 다양한 방식의 에이전트 제공 처리
    if isinstance(agents, BaseAgent):
        agents_dict = {"default": agents}
    elif isinstance(agents, list):
        agents_dict = {f"agent_{i}": agent for i, agent in enumerate(agents)}
    else:
        agents_dict = agents

    # 기본 에이전트 지정
    primary_key = data.get("primary_agent_key")
    if not primary_key and agents_dict:
        primary_key = next(iter(agents_dict))
        data["primary_agent_key"] = primary_key

    # 에이전트 딕셔너리 설정
    data["agents"] = agents_dict

    # BaseModel의 초기화 사용
    super().__init__(**data)

이 메서드는 다음과 같은 역할을 합니다:

• 단일 에이전트, 에이전트 리스트, 에이전트 딕셔너리 등 다양한 형태로 에이전트를 초기화할 수 있게 함
• 기본 에이전트를 자동으로 설정 (지정되지 않았을 경우 첫 번째 에이전트)
• Pydantic BaseModel의 초기화 활용

2. 에이전트 접근 메서드

@property
def primary_agent(self) -> Optional[BaseAgent]:
    """Get the primary agent for the flow"""
    return self.agents.get(self.primary_agent_key)

def get_agent(self, key: str) -> Optional[BaseAgent]:
    """Get a specific agent by key"""
    return self.agents.get(key)

def add_agent(self, key: str, agent: BaseAgent) -> None:
    """Add a new agent to the flow"""
    self.agents[key] = agent

이 메서드들은 다음과 같은 역할을 합니다:

• 기본 에이전트에 쉽게 접근할 수 있는 프로퍼티 제공
• 특정 에이전트를 키로 조회할 수 있는 기능
• 새로운 에이전트를 흐름에 추가하는 기능

3. 추상 실행 메서드

@abstractmethod
async def execute(self, input_text: str) -> str:
    """Execute the flow with given input"""

이 추상 메서드는 하위 클래스에서 구현해야 하며, 흐름의 실행 로직을 정의합니다.

PlanStepStatus 열거형 클래스 분석

PlanStepStatus는 계획 단계의 가능한 상태를 정의하는 열거형 클래스입니다.

class PlanStepStatus(str, Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    BLOCKED = "blocked"

1. 상태 값:

   • NOT_STARTED: 시작되지 않은 단계
   • IN_PROGRESS: 진행 중인 단계
   • COMPLETED: 완료된 단계
   • BLOCKED: 차단된 단계

2. 유틸리티 메서드:

   • get_all_statuses(): 모든 가능한 상태 값 반환
   • get_active_statuses(): 활성 상태 값 반환 (시작되지 않았거나 진행 중)
   • get_status_marks(): 상태에 대한 마커 기호 매핑 반환

설계 특징 및 패턴

1. 추상화: 추상 메서드를 통한 명확한 인터페이스 정의
2. 유연성: 다양한 방식으로 에이전트를 제공할 수 있는 유연한 초기화
3. 계층 구조: 기본 흐름 클래스를 확장하여 다양한 흐름 유형 구현 가능
4. 상태 관리: 계획 단계의 상태를 명확하게 정의하는 열거형
5. 다중 에이전트: 여러 에이전트를 지원하는 아키텍처
6. 비동기 처리: 모든 실행 메서드가 비동기(async)로 정의됨

이 BaseFlow 클래스는 OpenManus의 중요한 구성 요소로, 여러 에이전트를 조율하여 복잡한 작업을 수행하는 프레임워크를 제공합니다. 이 클래스를 상속받아 다양한 흐름 유형(예: 계획 흐름)을 구현할 수 있습니다.

10) app/flow/flow_factory.py

FlowFactory 클래스 분석

FlowFactory는 다양한 유형의 흐름을 생성하는 팩토리 클래스로, 여러 에이전트를 지원합니다.

주요 메서드

create_flow 정적 메서드

@staticmethod
def create_flow(
    flow_type: FlowType,
    agents: Union[BaseAgent, List[BaseAgent], Dict[str, BaseAgent]],
    **kwargs,
) -> BaseFlow:
    flows = {
        FlowType.PLANNING: PlanningFlow,
    }

    flow_class = flows.get(flow_type)
    if not flow_class:
        raise ValueError(f"Unknown flow type: {flow_type}")

    return flow_class(agents, **kwargs)

이 메서드는 다음과 같은 역할을 합니다:

1. 흐름 유형 매핑:

   • 흐름 유형을 해당 클래스에 매핑하는 딕셔너리를 정의
   • 현재는 FlowType.PLANNING만 PlanningFlow 클래스에 매핑됨

2. 유효성 검사:

   • 요청된 흐름 유형이 존재하는지 확인
   • 존재하지 않는 경우 ValueError 예외 발생

3. 유연한 에이전트 처리:

   • 단일 에이전트, 에이전트 목록 또는 에이전트 딕셔너리를 받을 수 있음
   • 이 유연성은 BaseFlow의 생성자에서 처리됨

4. 추가 매개변수 지원:

   • kwargs를 통해 추가 매개변수를 흐름 생성자에 전달

설계 특징 및 패턴

1. 팩토리 패턴: 객체 생성 로직을 클라이언트 코드에서 분리
2. 정적 메서드: 인스턴스 생성 없이 팩토리 메서드 사용 가능
3. 확장성: 새로운 흐름 유형을 쉽게 추가할 수 있는 구조
4. 유연성: 다양한 형태의 에이전트 입력 지원
5. 단일 책임: 객체 생성만을 담당하는 명확한 책임 분리

이 FlowFactory 클래스는 OpenManus 시스템에서 다양한 유형의 에이전트 흐름을 생성하기 위한 중앙화된 방법을 제공합니다. 현재는 PlanningFlow 유형만 지원하지만, 필요에 따라 새로운 흐름 유형을 쉽게 추가할 수 있도록 설계되어 있습니다.

11) app/flow/planning.py

PlanningFlow 클래스 분석

PlanningFlow는 에이전트를 사용하여 작업 계획을 수립하고 실행하는 흐름을 관리하는 클래스입니다.

주요 속성

llm: LLM = Field(default_factory=lambda: LLM())
planning_tool: PlanningTool = Field(default_factory=PlanningTool)
executor_keys: List[str] = Field(default_factory=list)
active_plan_id: str = Field(default_factory=lambda: f"plan_{int(time.time())}")
current_step_index: Optional[int] = None

1. LLM 및 계획 도구:

   • llm: 언어 모델 인스턴스
   • planning_tool: 계획 관리를 위한 도구

2. 실행 관리:

   • executor_keys: 단계 실행에 사용할 에이전트 키 목록
   • active_plan_id: 현재 활성화된 계획의 ID
   • current_step_index: 현재 실행 중인 단계의 인덱스

주요 메서드

1. 초기화 메서드

def __init__(
    self, agents: Union[BaseAgent, List[BaseAgent], Dict[str, BaseAgent]], **data
):
    # 설정 처리 및 부모 클래스 초기화
    # executor_keys 설정
    # planning_tool 초기화

이 메서드는 다음과 같은 역할을 합니다:

• 실행자 키를 설정(executors 매개변수 처리)
• 계획 ID 설정(제공된 경우)
• 계획 도구 초기화(제공되지 않은 경우)
• 모든 에이전트 키를 실행자 키로 사용(지정되지 않은 경우)

2. 실행자 선택 메서드

def get_executor(self, step_type: Optional[str] = None) -> BaseAgent:
    """단계 유형에 따라 적절한 실행자 에이전트를 선택"""
    # 단계 유형이 에이전트 키와 일치하면 해당 에이전트 사용
    # 그렇지 않으면 사용 가능한 첫 번째 실행자 또는 기본 에이전트 사용

이 메서드는 단계 유형에 따라 적절한 에이전트를 선택합니다:

• 단계 유형이 에이전트 키와 일치하면 해당 에이전트 사용
• 아니면 executor_keys 목록에서 첫 번째 사용 가능한 에이전트 사용
• 마지막으로 기본 에이전트로 폴백

3. 실행 메서드

async def execute(self, input_text: str) -> str:
    """에이전트를 사용하여 계획 흐름 실행"""
    # 입력이 제공되면 초기 계획 생성
    # 현재 단계 가져오기 및 실행
    # 모든 단계 완료 시 계획 마무리

이 메서드는 전체 계획 실행 흐름을 관리합니다:

• 기본 에이전트 확인
• 입력 텍스트에 기반한 초기 계획 생성
• 각 단계를 순차적으로 실행
• 종료 조건 처리(단계 없음 또는 에이전트 종료)
• 오류 처리

4. 초기 계획 생성 메서드

async def _create_initial_plan(self, request: str) -> None:
    """LLM과 PlanningTool을 사용하여 요청에 기반한 초기 계획 생성"""
    # 계획 생성을 위한 시스템 및 사용자 메시지 준비
    # LLM에 PlanningTool을 사용하여 계획 생성 요청
    # 도구 호출 처리 또는 기본 계획 생성

이 메서드는 초기 계획을 생성합니다:

• 계획 생성을 위한 프롬프트 준비
• 도구 선택으로 LLM 호출
• 반환된Ⅰ 도구 호출 처리
• 실패 시 기본 계획 생성

5. 현재 단계 정보 가져오기 메서드

async def _get_current_step_info(self) -> tuple[Optional[int], Optional[dict]]:
    """첫 번째 미완료 단계의 인덱스와 정보를 식별"""
    # 유효한 계획 확인
    # 단계 데이터 및 상태 가져오기
    # 첫 번째 활성 단계 찾기 및 상태 업데이트

이 메서드는 다음을 수행합니다:

• 계획 데이터 직접 접근
• 첫 번째 미완료 단계 찾기
• 단계 유형 추출(텍스트에서 패턴 매칭)
• 현재 단계를 'in_progress'로 표시

6. 단계 실행 메서드

async def _execute_step(self, executor: BaseAgent, step_info: dict) -> str:
    """지정된 에이전트를 사용하여 현재 단계 실행"""
    # 현재 계획 상태로 컨텍스트 준비
    # 에이전트에 단계 실행 프롬프트 제공
    # 단계 완료로 표시

이 메서드는 단계를 실행합니다:

• 계획 상태 포함한 컨텍스트 준비
• 적절한 프롬프트로 에이전트 실행
• 성공 시 단계를 완료로 표시

7. 계획 마무리 메서드

async def _finalize_plan(self) -> str:
    """계획을 마무리하고 LLM을 사용하여 요약 제공"""
    # 최종 계획 상태 가져오기
    # LLM을 사용하여 요약 생성
    # 실패 시 에이전트를 사용한 대체 요약

이 메서드는 계획을 마무리합니다:

• 최종 계획 텍스트 가져오기
• LLM을 사용하여 요약 생성
• 오류 시 에이전트를 사용한 폴백

설계 특징 및 패턴

1. 계층적 실행: 계획을 단계별로 순차적으로 실행
2. 동적 에이전트 선택: 단계 유형에 따른 적절한 에이전트 선택
3. 견고한 오류 처리: 각 단계에서 다양한 오류 상황에 대비
4. 직접/도구 대체 방식: 도구 호출 실패 시 직접 저장소 접근으로 대체
5. 상태 추적: 단계의 상태 추적 및 관리(시작되지 않음, 진행 중, 완료됨, 차단됨)
6. 유연한 요약: LLM 또는 에이전트를 통한 다중 요약 방법

이 PlanningFlow 클래스는 복잡한 작업을 계획하고 적절한 에이전트를 사용하여 단계별로 실행하는 강력한 프레임워크를 제공합니다. 각 단계마다 진행 상황을 추적하고 최종적으로 결과를 요약하는 흐름을 관리합니다.