|
|
from abc import ABC, abstractmethod |
|
|
from typing import Optional |
|
|
|
|
|
from pydantic import Field |
|
|
|
|
|
from app.agent.base import BaseAgent |
|
|
from app.llm import LLM |
|
|
from app.schema import AgentState, Memory |
|
|
|
|
|
|
|
|
class 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 |
|
|
|
|
|
@abstractmethod |
|
|
async def think(self) -> bool: |
|
|
"""Process current state and decide next action""" |
|
|
|
|
|
@abstractmethod |
|
|
async def act(self) -> str: |
|
|
"""Execute decided actions""" |
|
|
|
|
|
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() |
|
|
|
