Agents

Agents are the core unit of intelligence in Agentic Patterns. Each agent wraps an LLM and a system prompt, exposing a consistent interface for workflows and orchestrators to invoke.

BaseAgent

BaseAgent is the foundation all other agents build on. It takes an LLM and a system_prompt, builds a ChatPromptTemplate internally, and exposes a single invoke(user_input) method.

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.prompts import ChatPromptTemplate

class BaseAgent:
    def __init__(
        self,
        llm: BaseChatModel,
        system_prompt: str = "You are a helpful AI assistant.",
        agent_name: str = "BaseAgent"
    ):
        self.llm = llm
        self.system_prompt = system_prompt
        self.agent_name = agent_name
        self.prompt_template = ChatPromptTemplate.from_messages([
            ("system", self.system_prompt),
            ("human", "{input}")
        ])

    def invoke(self, user_input: str, **kwargs) -> str:
        chain = self.prompt_template | self.llm
        response = chain.invoke({"input": user_input, **kwargs})
        return response.content

The chain is assembled with | (LCEL pipe syntax): the template formats the messages, the LLM receives them, and response.content is returned as a plain string.

Specialized agents

PlanningAgent
ExecutionAgent
MonitoringAgent
LocalAgent

PlanningAgent breaks a high-level task into an ordered list of steps. Its system prompt instructs the LLM to return only a JSON object with a plan key.

from agents.planning_agent import PlanningAgent
from langchain_ollama import ChatOllama

llm = ChatOllama(model="llama3")
planner = PlanningAgent(llm=llm)

steps = planner.generate_plan("Research the capital of Andorra and summarize findings.")
# steps -> ["Step 1: Search Wikipedia for Andorra", "Step 2: Extract capital city", ...]

generate_plan() calls self.invoke(task), strips any markdown fences, then parses the JSON:

def generate_plan(self, task: str) -> List[str]:
    response_text = self.invoke(task)
    clean_text = response_text.replace("```json", "").replace("```", "").strip()
    parsed_data = json.loads(clean_text)
    return parsed_data.get("plan", [response_text])

If JSON parsing fails, the raw LLM response is returned as a single-element list so execution can still proceed.

ExecutionAgent carries out individual steps. When tools are supplied at construction time, it uses LangGraph’s create_react_agent to run a ReAct loop; otherwise it falls back to the plain invoke() path.

from agents.execution_agent import ExecutionAgent
from tools.curl_search_tool import CurlSearchTool

curl_tool = CurlSearchTool()
executor = ExecutionAgent(llm=llm, tools=[curl_tool])

result = executor.execute_step(
    step_description="Search Wikipedia for the capital of Andorra",
    context="Previous steps found that Andorra is a small European principality."
)

The context parameter carries accumulated results from earlier steps:

def execute_step(self, step_description: str, context: str = "") -> str:
    input_prompt = f"Step to Execute: {step_description}"
    if context:
        input_prompt += f"\n\nContext:\n{context}"

    if self.agent_executor:
        full_prompt = f"{self.system_prompt}\n\n{input_prompt}"
        response = self.agent_executor.invoke({"messages": [("user", full_prompt)]})
        return response["messages"][-1].content

    return self.invoke(input_prompt)

When tools are provided, ExecutionAgent prepends the system prompt directly into the user message. This ensures the instruction is respected across all LangGraph versions that may not support state_modifier.

MonitoringAgent evaluates whether a step’s output actually satisfied its objective. Its system prompt demands a JSON response with success (bool) and feedback (str).

from agents.monitoring_agent import MonitoringAgent

monitor = MonitoringAgent(llm=llm)

evaluation = monitor.evaluate(
    objective="Search Wikipedia for the capital of Andorra",
    result="According to Wikipedia, Andorra la Vella is the capital."
)
# evaluation -> {"success": True, "feedback": "Objective met completely."}

The method builds an evaluation prompt, invokes the LLM, and parses the JSON:

def evaluate(self, objective: str, result: str) -> Dict[str, Any]:
    evaluation_prompt = (
        f"Original Objective:\n{objective}\n\n"
        f"Actual Output:\n{result}\n\n"
        "Evaluate if the Actual Output successfully meets the Original Objective."
    )
    response_text = self.invoke(evaluation_prompt)
    clean_text = response_text.replace("```json", "").replace("```", "").strip()
    parsed_data = json.loads(clean_text)
    return {
        "success": bool(parsed_data.get("success", False)),
        "feedback": str(parsed_data.get("feedback", "No feedback provided."))
    }

On JSON parse failure, evaluate() returns {"success": False, "feedback": ...} so the workflow can retry rather than silently pass a bad result.

LocalAgent is a purpose-built summarization agent intended to run against a local model (e.g., via Ollama). It has the same invoke() interface as BaseAgent but is not a subclass — it is a standalone class suited for context compression.

from agents.local_agent import LocalAgent
from langchain_ollama import ChatOllama

local_llm = ChatOllama(model="llama3")
compressor = LocalAgent(llm=local_llm)

condensed = compressor.invoke(
    "[Step 1 Result]: Andorra la Vella is the capital and largest city..."
)
# condensed -> much shorter representation preserving key facts

Its default system prompt instructs the model to “condense the provided text aggressively to save tokens while keeping all crucial facts.”

Use LocalAgent as the compressor argument to LangGraphOrchestrator or as tools[0] in SequentialWorkflow when Ollama is available. Fall back to CompressContextTool when it is not.

The Plan-Execute-Monitor loop

All three specialized agents participate in a recurring feedback loop that drives multi-step task completion:

Plan

PlanningAgent.generate_plan(task) converts a high-level objective into an ordered List[str] of step descriptions.

Execute

ExecutionAgent.execute_step(step, context) carries out the current step, optionally using tools via a ReAct loop. Accumulated results from previous steps are passed as context.

Monitor

MonitoringAgent.evaluate(objective, result) checks whether the step’s output satisfied its objective and returns structured {success, feedback}.

Retry or advance

On failure, feedback is appended to the context and the executor retries (up to max_retries). On success, the result is appended to context and the loop advances to the next step.

This loop is implemented by both SequentialWorkflow and LangGraphOrchestrator — see Workflows and Orchestration for details on how each coordinates these agents.

Agent summary

BaseAgent

Foundation class. Takes llm + system_prompt. Exposes invoke(user_input). Uses ChatPromptTemplate + LCEL internally.

PlanningAgent

Extends BaseAgent. generate_plan(task) returns List[str] by parsing JSON {"plan": [...]} from the LLM.

ExecutionAgent

Extends BaseAgent. execute_step(step, context) runs steps with optional tools via create_react_agent.

MonitoringAgent

Extends BaseAgent. evaluate(objective, result) returns {success: bool, feedback: str} parsed from JSON.

LocalAgent

Standalone summarization agent. Same invoke() interface as BaseAgent. Designed for local LLMs (Ollama).

Get Started

Core Concepts

Guides

BaseAgent

Specialized agents

The Plan-Execute-Monitor loop

Agent summary

BaseAgent

PlanningAgent

ExecutionAgent

MonitoringAgent

LocalAgent

Build docs developers (and LLMs) love

Get Started

Core Concepts

Guides

​BaseAgent

​Specialized agents

​The Plan-Execute-Monitor loop

​Agent summary

BaseAgent

PlanningAgent

ExecutionAgent

MonitoringAgent

LocalAgent

Build docs developers (and LLMs) love

BaseAgent

Specialized agents

The Plan-Execute-Monitor loop

Agent summary