Building an LLM-Powered Autonomous Agent Node

Agent Node Implementation - LLM-Based Autonomous Execution System

The agent node was one of the most complex features to build on the XGen platform. This post covers how it was implemented. An agent node is an autonomous execution system in which the LLM decides on its own which tools to call, then iteratively acts on the results.

Agent Architecture

The agent is designed around the ReAct (Reasoning + Acting) pattern.

1. The LLM analyzes the current state and decides the next action
2. It calls a Tool to perform the task
3. It observes the result and re-evaluates
4. When the goal is reached, it returns a final response

Tool Definition

from typing import Callable, Any
from pydantic import BaseModel

class ToolDefinition(BaseModel):
    name: str
    description: str
    parameters: dict
    function: Callable

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, ToolDefinition] = {}

    def register(self, name: str, description: str, parameters: dict):
        def decorator(func: Callable):
            self._tools[name] = ToolDefinition(
                name=name,
                description=description,
                parameters=parameters,
                function=func,
            )
            return func
        return decorator

    def get_openai_tools(self) -> list:
        return [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.parameters,
            },
        } for tool in self._tools.values()]

tools = ToolRegistry()

@tools.register(
    name="search_documents",
    description="벡터 DB에서 관련 문서를 검색합니다",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "검색 쿼리"},
            "top_k": {"type": "integer", "default": 5},
        },
        "required": ["query"],
    },
)
async def search_documents(query: str, top_k: int = 5):
    return await vector_search(query, top_k)

Agent Execution Loop

class AgentNode(BaseNode):
    MAX_ITERATIONS = 10

    async def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        messages = [
            {"role": "system", "content": self.config.params.get(
                "system_prompt",
                "당신은 도구를 활용하여 작업을 수행하는 에이전트입니다."
            )},
            {"role": "user", "content": inputs.get("input", "")},
        ]

        for iteration in range(self.MAX_ITERATIONS):
            response = await self.client.chat.completions.create(
                model=self.config.params.get("model", "gpt-4o"),
                messages=messages,
                tools=tools.get_openai_tools(),
                tool_choice="auto",
            )

            message = response.choices[0].message
            messages.append(message)

            # No tool calls means this is the final response
            if not message.tool_calls:
                return {"output": message.content}

            # Execute tools
            for tool_call in message.tool_calls:
                func = tools._tools[tool_call.function.name].function
                args = json.loads(tool_call.function.arguments)
                result = await func(**args)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result, ensure_ascii=False),
                })

        return {"output": "최대 반복 횟수에 도달했습니다.", "iterations": self.MAX_ITERATIONS}

Safeguards

Several safeguards were implemented to prevent the agent from getting stuck in infinite loops or invoking unexpected tools.

class AgentSafeguard:
    def __init__(self, max_iterations: int = 10, max_tokens: int = 50000):
        self.max_iterations = max_iterations
        self.max_tokens = max_tokens
        self.total_tokens = 0

    def check_budget(self, usage) -> bool:
        self.total_tokens += usage.total_tokens
        if self.total_tokens > self.max_tokens:
            raise AgentBudgetExceeded(
                f"토큰 예산 초과: {self.total_tokens}/{self.max_tokens}"
            )
        return True

The agent node goes well beyond a simple LLM call — it enables complex, multi-step tasks to be executed automatically. The biggest win was being able to handle pipelines like RAG retrieval → result analysis → follow-up retrieval → final answer generation, all within a single node.