跳到正文
DEV Community· Tamiz Uddin·· 4 小時前AI 評分40

由人工照看走向自主:打造可自行修正、無須人工介入的 AI 智能體

From Babysitting to Autonomy: Engineering AI Agents That Self-Correct Without Human Intervention

AI 導讀

文章探討如何透過工程架構,打造可自行修正、無須人工介入的 AI 智能體。做法包括以明確的狀態機管理智能體、拒絕違規操作並回饋原因,以及由 Critic 評估輸出,配合結構化驗證提高可靠性。

正文

Originally published on tamiz.pro.

Most Large Language Model (LLM) applications currently deployed in production are not autonomous; they are fragile dependencies wrapped in HTTP calls that break the moment the model outputs an unexpected JSON key or fails to follow a multi-step instruction set. Engineers are spending the majority of their sprint capacity "babysitting" these models—manually injecting context, retrying failed generations, and hard-coding exception handlers for every possible hallucination. The result is brittle systems that require constant human supervision, effectively turning an AI-powered feature into a high-maintenance manual workflow.

True autonomy in AI systems does not mean letting the model act without constraints. It means building a deterministic scaffolding around the non-deterministic model so that errors are caught, corrected, or contained before they propagate. This requires a shift in paradigm: moving from treating the LLM as a magical oracle to treating it as a probabilistic component in a larger, verifiable state machine.

In this deep-dive, we will explore the architectural patterns required to eliminate the need for human supervision. We will examine how to structure prompts for self-correction, implement output validation layers that force the model into a known schema, and design feedback loops that allow the agent to retry its own actions. By adopting these engineering practices, you can transition your AI tools from "assisted" to "autonomous," reducing operational overhead and unlocking true scalability.

Table of Contents

1. The Cost of Fragility: Why Human Supervision Scales Poorly

The fundamental problem with "babysitting" an AI agent is that human attention is the bottleneck. In a support-ticket classification system, if the AI gets a ticket wrong 5% of the time, a human can review the 100 failed tickets out of 2,000 daily. This is manageable. But if the system processes 20,000 tickets a day, that 5% error rate becomes 1,000 tickets requiring human review. The system has not failed, but the workflow has failed because it cannot scale.

Fragility in LLM applications usually stems from two sources:

  1. Output Variance: The model might return valid JSON in one call and a markdown block in the next.
  2. Context Drift: Over multiple turns, the model loses sight of the original goal, leading to actions that are locally logical but globally incorrect.

To move toward autonomy, we must stop relying on the model's intent and start relying on verifiable outcomes. If a human cannot verify the outcome, the system cannot be autonomous. If a machine can verify the outcome, the human can be removed from the loop.

2. The Architecture of Autonomy: Defining the Agent’s Boundary

An autonomous agent is not just a prompt; it is a state machine. The first step in removing human supervision is to define the agent's state explicitly. Instead of passing a conversation history to the LLM and hoping it stays on track, we maintain a structured state object.

Consider the following state structure for a data retrieval agent:

class AgentState:
    goal: str
    current_action: str
    last_observation: str
    error_log: list
    success: bool = False

The LLM’s job is not to "think" in a vacuum; its job is to map current_state to next_action. This architectural shift allows us to implement a Rejection Sampling pattern. If the LLM proposes an action that violates our constraints, we do not wait for a human to fix it. We immediately reject it, append the rejection reason to error_log, and feed the state back to the LLM. The model learns, in real-time, what is not allowed, without human intervention.

3. Self-Correction Loops: Teaching the Model to Debug Itself

Most autonomous systems fail because they trust the first output. Autonomy requires a feedback loop where the model critiques its own work. This is often referred to as Reflexion or Self-Consistency.

The Critic-Evaluator Pattern

Instead of a single LLM call, we use two roles:

  1. The Actor: Generates the solution.
  2. The Critic: Evaluates the solution against a rubric.
def self_correct(task: str, initial_solution: str, rubric: str) -> str:
    critic_prompt = f"""
    You are a strict code reviewer.
    Task: {task}
    Proposed Solution: {initial_solution}

    Review the solution against the following rubric:
    {rubric}

    If the solution is perfect, return "ACCEPT".
    If it has flaws, return specific feedback starting with "REJECT: [reasons]".
    """

    critique = llm.generate(critic_prompt)

    if critique.startswith("ACCEPT"):
        return initial_solution

    # Feed the critique back to the Actor to generate a revised solution
    revision_prompt = f"""
    Task: {task}
    Initial Attempt: {initial_solution}
    Critic Feedback: {critique}

    Revise the solution to address all points in the feedback.
    """
    return llm.generate(revision_prompt)

This loop can be run up to N times. If the Critic rejects the solution three times, the agent can escalate to a higher-level model or trigger a fallback. Crucially, this entire process happens without a human looking at the intermediate states. The "supervision" is encoded in the rubric.

4. Structural Constraints: Forcing Compliance via Schema Validation

Autonomy breaks down when the output format is unpredictable. To eliminate the need for humans to parse messy text, we must enforce structural constraints.

JSON Mode and Pydantic

Modern LLM providers support "JSON Mode," which constrains the output to valid JSON. However, valid JSON is not enough; we need typed JSON. By using schema validation libraries like Pydantic (Python) or Zod (TypeScript), we can catch errors immediately.

import { z } from "zod";
import { openai } from "@ai-sdk/openai";
import { generateObject } from "ai";

// 1. Define the strict schema
const AgentActionSchema = z.object({
    tool_name: z.enum(["search_db", "run_script", "finish"]),
    arguments: z.record(z.string()),
    reasoning: z.string(),
});

// 2. Use the AI SDK's generateObject which enforces the schema
const { object, usage } = await generateObject({
    model: openai("gpt-4o"),
    schema: AgentActionSchema,
    prompt: "Determine the next step to solve the user's query: 'Find all users who signed up last week'.",
});

// object is now a strongly typed instance of AgentActionSchema
// No manual JSON parsing or try-catch blocks needed for format errors.

If the model hallucinates a tool named "delete_all" instead of "run_script", the Zod validation fails instantly. The agent can then catch this error, log it, and retry the generation with the validation error message included in the prompt. This turns a potential catastrophic failure into a recoverable loop.

5. The ReAct Pattern: Reasoning and Acting in Concert

The most robust pattern for autonomous agents is ReAct (Reasoning + Acting). It interleaves natural language thoughts with specific tool calls. This prevents the "blind confidence" problem where a model thinks it has solved a problem but actually hasn't executed the necessary steps.

Implementing the Loop

The ReAct loop is a while loop that continues until a Finish action is taken or a limit is reached.

import json

def react_loop(task: str, max_steps: int = 5):
    history = []
    for i in range(max_steps):
        # 1. Reasoning
        prompt = f"""
        Goal: {task}
        History: {json.dumps(history)}

        Format:
        Thought: <your reasoning>
        Action: <JSON object with tool_name and arguments>
        """

        response = llm.generate(prompt)
        # Parse response into Thought and Action
        thought, action_json = parse_response(response)
        history.append({"thought": thought, "action": action_json})

        # 2. Acting
        if action_json.get("tool_name") == "finish":
            return action_json.get("final_answer")

        tool_output = execute_tool(action_json)

        # 3. Observing
        history.append({"observation": tool_output})

    return "Max steps reached without resolution"

In this pattern, the "Observation" is critical. If execute_tool returns an error (e.g., "File not found"), that error becomes part of the history. The LLM sees the error in the next iteration and can adjust its Thought ("The file doesn't exist, I need to create it first"). This self-correction is the core of autonomy: the agent learns from its own failures without human input.

6. Safety Rails and Kill Switches: Autonomous Without Dangerous

Autonomy without safety is a liability. We must implement deterministic "hard stops" that the LLM cannot override.

Budgeting and Limits

  1. Token Budget: Track the cumulative token usage. If the agent exceeds a threshold (e.g., 50,000 tokens), halt the loop to prevent cost overruns.
  2. Action Whitelist: The AgentActionSchema should only allow safe tools. Dangerous tools (like rm -rf or DROP TABLE) should not be in the enum, or should require a secondary verification step.
  3. Deterministic Guards: Before executing any tool, run a deterministic check.

    def safe_execute(action: dict) -> str:
        if action["tool_name"] == "db_write":
            if not action["arguments"]["transaction"]: 
                return "ERROR: Write operations must be transactional"
        # Execute only if safe
        return execute(action)
    

These guards act as the "immune system" of the agent. They don't understand context, but they enforce invariants. This allows the LLM to be creative in reasoning while being rigid in action, removing the need for a human to double-check every SQL query.

7. Implementation: A Robust Autonomous Coding Agent

Let’s synthesize these concepts into a practical example: an agent that writes a Python function, runs it, and fixes bugs until it passes a test suite.

The Workflow

  1. Prompt: "Write a function parse_log that handles edge case X."
  2. Generate: LLM produces code.
  3. Validate: Run unit tests.
  4. Feedback: If tests fail, pass the traceback to the LLM.
  5. Iterate: LLM modifies code.
  6. Repeat: Until tests pass or max retries hit.

Code Snippet

import subprocess

def autonomous_coder(task: str, test_cmd: str, max_attempts: int = 3):
    code = """
    def parse_log(line: str) -> dict:
        # Initial guess
        pass
    """

    for attempt in range(max_attempts):
        # 1. Write code to file
        with open("solution.py", "w") as f:
            f.write(code)

        # 2. Run tests
        result = subprocess.run(["python", test_cmd], capture_output=True, text=True)

        if result.returncode == 0:
            return code

        # 3. If failed, ask LLM to fix based on stderr
        prompt = f"""
        Task: {task}
        Current Code: {code}
        Test Failure Output:
        {result.stderr}

        Provide the corrected code only.
        """
        code = llm.generate(prompt)

    raise Exception("Agent failed to resolve bug")

This pattern eliminates the "babysitter." A human is not needed to look at the test failure and type out the fix. The agent closes the loop autonomously. The only human intervention is the initial task definition and the final verification of the passing test.

8. Frequently Asked Questions

How do I prevent the agent from going into an infinite loop?
Always implement a max_steps counter. In the ReAct loop above, if i reaches max_steps, the agent must stop and return its best-effort result or an error state. Additionally, monitor the diversity of actions; if the agent keeps trying the same failed tool, inject a meta-prompt instructing it to try a different approach or admit failure.

Is it better to use a smaller, faster model for the "Critic"?
Yes. The "Actor" model (generating the code/logic) can be a large, reasoning-heavy model (like GPT-4 or Claude 3 Opus). The "Critic" model (checking syntax, basic logic, or format) can be a smaller, faster model (like GPT-3.5 or a local Llama 3 8B). This hybrid approach reduces latency and cost while maintaining high quality for the complex parts of the task.

How do I handle long-running tasks that exceed context limits?
Implement a "Memory Summarization

Pipeline to manage this. As an agent processes a multi-step workflow, it generates a running log of actions and observations. At regular intervals—such as every five steps, or when the context window approaches 80% capacity—the system triggers a summarization routine. A dedicated, smaller model (often a fast inference model like Mistral 7B or a specialized quantized variant) reads the accumulated context and distills it into a concise state object. This object contains:

  1. Goal: The current primary objective.
  2. Constraints: Any newly discovered limitations or rules.
  3. Progress: What has been successfully completed.
  4. Open Issues: Unresolved errors or pending validations.

The raw history is then archived to long-term storage (vector databases or SQL), and the agent’s working context is replaced with this distilled state. This allows the agent to "forget" the noise of intermediate steps while retaining the semantic essence of its journey.

Code Implementation: Hierarchical Memory Management

Below is a practical example using LangChain’s ConversationSummaryBufferMemory combined with a custom vector store for retrieval augmentation. This demonstrates how to offload older context and retrieve relevant details only when needed.

from langchain_community.llms import OpenAI
from langchain.chains.conversation_memory import ConversationTokenBufferMemory
from langchain.memory import ConversationTokenBufferMemory
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.document_compressors import LLMChain
from langchain.prompts.chat import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)
from langchain_community.chat_models import ChatOpenAI
from typing import List, Dict, Any

class SelfCorrectingAgentMemory:
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = OpenAI(model_name=model_name, temperature=0)
        self.embedding_model = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
        self.vector_store = Chroma(embedding_function=self.embedding_model, persist_directory="./chroma_db")

        # Initialize token buffer memory with a large limit
        self.memory = ConversationTokenBufferMemory(
            llm=self.llm,
            max_token_limit=4000,
            input_key="input",
            output_key="output"
        )

        # Summarization prompt for overflow handling
        self.summarization_prompt = ChatPromptTemplate.from_messages([
            SystemMessagePromptTemplate.from_template(
                "You are a memory manager. Summarize the conversation below to retain only critical context. "
                "Identify the current goal, completed steps, and open issues. Use bullet points."
            ),
            HumanMessagePromptTemplate.from_template("{messages}")
        ])

    def add_interaction(self, user_input: str, agent_output: str):
        """Adds new interactions and manages overflow via summarization."""
        self.memory.save_context({"input": user_input}, {"output": agent_output})

        # Check if we've exceeded our threshold
        if len(self.memory.buffer_as_str()) > 3500:
            self._summarize_and_offload()

    def _summarize_and_offload(self):
        """Summarizes current buffer and stores details in vector DB."""
        current_context = self.memory.buffer_as_str()

        # 1. Generate Summary
        summary_llm_chain = LLMChain(
            llm=self.llm,
            prompt=self.summarization_prompt,
            output_key="summary"
        )

        summary = summary_llm_chain.run(messages=current_context)

        # 2. Offload raw details to Vector DB
        # We chunk the raw context for retrieval
        docs = [f"Step {i}: {line}" for i, line in enumerate(current_context.split('\n'))]
        self.vector_store.add_texts(docs)

        # 3. Reset Memory to Summary only
        self.memory.clear()
        self.memory.buffer = f"System Summary:\n{summary}"

    def retrieve_relevant_context(self, query: str, k: int = 3) -> str:
        """Retrieves specific past details if the summary is insufficient."""
        relevant_docs = self.vector_store.similarity_search(query, k=k)
        if not relevant_docs:
            return ""
        return "\n".join([doc.page_content for doc in relevant_docs])

# Usage Example
memory_manager = SelfCorrectingAgentMemory()

# Simulating a long task
memory_manager.add_interaction("Install PostgreSQL and set up schema", "Completed installation. Schema created.")
memory_manager.add_interaction("Insert test data into users table", "Failed: Constraint violation on email uniqueness.")
memory_manager.add_interaction("Retry with unique emails", "Success. 500 records inserted.")

# When agent needs to verify if 'users' table was modified recently
if "users" in memory_manager.memory.buffer:
    detailed_history = memory_manager.retrieve_relevant_context("users table modifications")
    print(f"Retrieved Context:\n{detailed_history}")

The Self-Correction Loop: Reflection and Retry

The most critical component of autonomous agents is the ability to detect failure and adjust strategy. This is not just about catching exceptions; it is about semantic verification. An agent might technically complete a code execution without throwing an error, but the output might still be logically incorrect.

The Reflexion Framework

Shinn et al.’s Reflexion framework popularized the idea of "verbal reinforcement." Instead of updating weights (which is impractical for real-time inference), the agent updates its prompt based on a self-generated critique.

  1. Action: The agent executes a step (e.g., writes a SQL query).
  2. Evaluation: A separate critic model (or the same model with a different system prompt) evaluates the result against the goal.
  3. Reflection: If the evaluation fails, the agent generates a natural language reflection: "The query returned 0 rows because I used = instead of LIKE for the search term. Next time, I should use partial matching."
  4. Retry: The agent is re-prompted with the original task plus the reflection text.

Implementing a Critic-Evaluator Pattern

Here, we implement a robust retry mechanism that distinguishes between transient errors (network timeouts) and logical errors (bad code).

import time
from typing import Callable, Any

class AgentRetryStrategy:
    def __init__(self, max_retries: int = 3, backoff_factor: float = 1.5):
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.history: List[str] = []

    def execute_with_reflection(self, task_fn: Callable[[], Any], critic_fn: Callable[[Any], Dict]) -> Any:
        """
        Executes a task and uses a critic to determine if it succeeded.
        If it fails, it generates a reflection and retries.
        """
        last_result = None
        reflection_context = ""

        for attempt in range(self.max_retries):
            # 1. Execute the task with current reflection context
            try:
                result = task_fn(reflection_context)
            except Exception as e:
                # Transient error handling
                if "timeout" in str(e).lower():
                    time.sleep(self.backoff_factor ** attempt)
                    continue
                else:
                    last_result = str(e)
                    reflection_context = self._generate_critique(f"Exception: {e}", last_result)
                    continue

            # 2. Critic evaluates the result
            critique = critic_fn(result)

            if critique["success"]:
                return result

            # 3. Failure - Generate Reflection
            last_result = result
            reflection_context = self._generate_critique(critique["reason"], result)
            self.history.append(reflection_context)

        raise Exception(f"Task failed after {self.max_retries} attempts. Last error: {last_result}")

    def _generate_critique(self, reason: str, result: Any) -> str:
        """
        In a real system, this calls an LLM to generate a natural language reflection.
        Here we simulate a simple heuristic for demonstration.
        """
        # Pseudo-code for LLM call:
        # prompt = f"Task failed because: {reason}. Current result: {result}. 
        #          What was the mistake? How should the next attempt differ?"
        # return llm.generate(prompt)

        return f"Previous attempt failed due to: {reason}. Ensure that all edge cases are handled and re-verify assumptions before executing."

Autonomous Tool Use and Dynamic Planning

Static planning (pre-defined steps) fails when environments are dynamic. Autonomous agents must plan on the fly. This requires a ReAct (Reasoning + Acting) loop where the agent interleaves thoughts and actions.

The ReAct Loop in Production

In production environments, the ReAct loop must be robust against infinite loops. We introduce a "Step Budget" and a "Diversity Check."

  1. Step Budget: The agent is allowed a maximum number of tool calls (e.g., 10). If it exceeds this, it must halt and report partial progress.
  2. Diversity Check: The system monitors the agent's recent actions. If the agent attempts the same action with the same parameters three times, the system flags a "stall" and forces a change in strategy (e.g., switching from "Try hard" to "Ask for clarification" or "Use a different tool").

Code: A Robust ReAct Agent with Stall Detection

from dataclasses import dataclass
from typing import Union, List

@dataclass
class AgentState:
    thought: str
    action: str
    observation: str
    step_count: int = 0
    last_action_hash: str = ""

class RobustReActAgent:
    def __init__(self, llm, tools: List[str], max_steps: int = 10):
        self.llm = llm
        self.tools = tools
        self.max_steps = max_steps
        self.state = AgentState("", "", "", 0)
        self.action_history: List[str] = []

    def run(self, query: str) -> str:
        context = self._build_context(query)

        while self.state.step_count < self.max_steps:
            self.state.step_count += 1

            # 1. Reasoning
            response = self._reason(context)
            thought, action = self._parse_response(response)

            # 2. Stall Detection
            action_hash = self._hash(action)
            if action_hash == self.state.last_action_hash:
                if self.action_history.count(action_hash) >= 2:
                    context += f"\nWARNING: You have repeated the same action twice. You must choose a different tool or parameter set.\n"
                    # Force a diverse response
                    response = self._reason(context + "Please try a completely different approach.")
                    thought, action = self._parse_response(response)

            self.state.last_action_hash = action_hash
            self.action_history.append(action_hash)

            # 3. Execution
            if action.startswith("FINISH:"):
                return action.split("FINISH:")[1].strip()

            observation = self._execute_action(action)

            # 4. Update Context
            context = self._update_context(context, thought, action, observation)

        return "ERROR: Max steps exceeded. Please review the partial progress in the observation log."

    def _reason(self, context: str) -> str:
        """
        This is where the LLM is called.
        Prompt structure:
        Context: {context}
        Thought: ...
        Action: ...
        """
        # Simulated LLM call
        # In reality: return self.llm.generate(context)
        return "Thought: I need to search. Action: search[python async]"

    def _parse_response(self, response: str) -> tuple:
        # Simplified parser for demonstration
        lines = response.split('\n')
        thought = lines[0].replace("Thought:", "").strip() if lines else ""
        action = lines[1].replace("Action:", "").strip() if len(lines) > 1 else ""
        return thought, action

    def _execute_action(self, action: str) -> str:
        # Simulated tool execution
        return f"Observation: Found 5 relevant documents about {action.split('[')[-1].rstrip(']}')"

    def _update_context(self, context: str, thought: str, action: str, observation: str) -> str:
        return context + f"\nThought: {thought}\nAction: {action}\nObservation: {observation}"

    def _hash(self, action: str) -> str:
        return hash(action)

Handling Hallucinations in Agent Outputs

Autonomy increases the risk of compounding errors. If Step 1 hallucinates a variable name, Step 2 will likely fail, and the agent may hallucinate a "fix" that is entirely bogus.

Grounding Techniques

To mitigate this, employ Grounded Verification:

  1. Source Citation: Force the agent to cite specific documents or API responses for every claim in its reasoning. If a citation cannot be verified against the original data, the step is rejected.
  2. Code Execution as Truth: For programming tasks, never trust the LLM's self-assessment of code correctness. Always execute the code in a sandboxed environment. If it throws an error, feed the exact stack trace back to the agent.
  3. Cross-Validation: Use two different agents or models to verify each other. Agent A proposes a solution; Agent B (with a critical persona) attempts to break it. Only if B fails to break it is the solution accepted.

Code: Sandboxed Execution for Code Agents

import subprocess
import tempfile
import os

def execute_code_safely(code: str, timeout: int = 5) -> Dict[str, Any]:
    """
    Executes Python code in a temporary file with a strict timeout.
    Returns output and success status.
    """
    with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as f:
        f.write(code.encode('utf-8'))
        temp_path = f.name

    try:
        result = subprocess.run(
            ["python", temp_path],
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return {
            "success": result.returncode == 0,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except subprocess.TimeoutExpired:
        return {"success": False, "stdout": "", "stderr": "Timeout exceeded"}
    finally:
        os.unlink(temp_path)

# Usage in Agent Loop
# if not result["success"]:
#     agent_feedback = f"Code execution failed. Error: {result['stderr']}. Fix the syntax error and retry."

Conclusion: The Path to Trustworthy Autonomy

Building AI agents that can operate without human intervention is not just about prompting models; it is about engineering systems that can fail safely and recover intelligently. The key pillars are:

  1. Robust Memory Management: Using hierarchical memory to handle long-context tasks without degrading performance.
  2. Self-Correction Loops: Implementing reflection and critic patterns to allow agents to learn from their own mistakes in real-time.
  3. Grounded Execution: Treating code execution and data retrieval as the source of truth, rather than relying on the LLM's internal knowledge.
  4. Guardrails and Budgets: Preventing infinite loops and resource exhaustion through step limits and stall detection.

As we move forward, the focus will shift from "Can the agent do the task?" to "Can the agent prove it did the task correctly?" The next generation of autonomous systems will be defined by their ability to provide verifiable, auditable trails of their reasoning, allowing humans to intervene only when necessary, rather than constantly. The goal is not to replace humans, but to amplify human intent with reliable, self-correcting digital workhorses.

來源:DEV Community · dev.to