Project Guide
Build a Self-Correcting Coding Agent with LangGraph and E2B
Source: mortalapps.comThis guide provides a comprehensive tutorial on how to build a robust and self-correcting coding agent python build using LangGraph and E2B. The agent is designed to understand a given problem, generate Python code, execute it in a secure sandbox, analyze the output, and iteratively refine its code until the problem is solved. This project is ideal for AI engineers, software developers, and researchers looking to implement advanced autonomous agents capable of complex problem-solving and code generation.
The core challenge in building coding agents is ensuring reliability and correctness. Traditional approaches often struggle with hallucinated code or execution errors. This guide addresses these issues by integrating a test-driven development loop, leveraging E2B for secure and reproducible code execution, and implementing a reflection mechanism that allows the agent to learn from failures and self-correct. By the end of this project, you will have a deployable system that can tackle various Python coding challenges autonomously.
Readers will gain practical experience with state machine design using LangGraph, secure tool execution with sandboxing, and incorporating episodic memory for improved performance. This architecture offers a significant advantage over simpler prompt-response agents by enabling dynamic problem-solving, reducing manual intervention, and fostering a more resilient and intelligent agent capable of complex software development tasks.
What You Will Build
You will build a Python-based coding agent that operates within a LangGraph state machine. This agent will take a natural language problem description as input, generate Python code to solve it, and execute that code in an isolated E2B sandbox. The agent will then analyze the execution results, including stdout, stderr, and any errors. If the code fails or doesn't meet the requirements, the agent will reflect on the failure, identify the root cause, and generate corrected code, repeating this cycle until a successful solution is found or a retry limit is reached.
The agent communicates primarily through text prompts and code snippets. Its tools include a large language model for generation and reflection, and an E2B sandbox for code execution. The system will run as a Python application, designed for easy containerization and deployment. The output of the agent will be the final, validated Python code solution along with its successful execution output. The final architecture comprises a central LangGraph orchestrator managing nodes for code generation, execution, output parsing, and self-reflection, all leveraging a secure E2B sandbox for safe code execution and an in-memory episodic memory for storing successful solutions.
Technology Stack
| Component | Technology | Why This Choice |
|---|---|---|
| LLM Provider | OpenAI (GPT-4o) |
Provides state-of-the-art code generation, reasoning, and reflection capabilities crucial for a coding agent. |
| Orchestration Framework | LangGraph |
Enables the definition of complex, stateful agent workflows with conditional routing, cycles, and persistent state, ideal for iterative code generation and correction. |
| Code Execution Sandbox | E2B |
Offers a secure, isolated, and reproducible environment for executing generated code, preventing malicious code from affecting the host system and providing detailed execution feedback. |
| Memory | In-memory dictionary (Episodic) |
For this project, a simple in-memory dictionary stores successful solutions as a basic form of episodic memory, demonstrating the concept without external dependencies. This can be easily upgraded later. |
| Observability | Python logging, basic print statements |
Provides essential visibility into the agent's decision-making process, tool calls, and execution results for debugging and understanding agent behavior during development. |
| Authentication & Secrets | Environment Variables |
Standard and secure practice for managing API keys and other sensitive configuration, ensuring they are not hardcoded in the codebase. |
| Deployment | Docker, Kubernetes (example) |
Containerization with Docker provides a reproducible deployment environment, while Kubernetes offers a scalable and resilient orchestration platform for production. |
The choice of OpenAI's GPT-4o as the Large Language Model (LLM) provider is driven by its exceptional code generation, understanding, and reasoning capabilities. For a coding agent that needs to not only write but also debug and reflect on code, GPT-4o offers the necessary intelligence and contextual awareness to perform complex tasks effectively. While other models like Claude 3 Opus or Google's Gemini could also be viable, GPT-4o provides a strong balance of performance and accessibility for this guide.
LangGraph is selected as the orchestration framework due to its powerful state machine capabilities. Unlike simpler sequential frameworks, LangGraph allows for the creation of intricate, cyclical workflows that are essential for a self-correcting agent. The ability to define nodes for distinct actions (generate, execute, reflect) and edges for conditional transitions makes it perfectly suited for the iterative nature of code development. Alternatives like CrewAI or custom state management could work, but LangGraph provides a robust and well-supported solution specifically designed for agentic workflows.
For secure code execution, E2B is the chosen sandbox environment. Running arbitrary, LLM-generated code directly on the host system is a significant security risk. E2B provides isolated, ephemeral environments that execute code safely and return detailed outputs, including stdout, stderr, and error messages. This granular feedback is critical for the agent's reflection mechanism. While other sandbox solutions exist, E2B offers a developer-friendly API and robust features for this use case.
Episodic memory is initially implemented using a simple in-memory dictionary. This choice prioritizes clarity and ease of setup for a tutorial, allowing readers to focus on the core agent logic without the overhead of setting up a separate vector database. This is a deliberate simplification to demonstrate the concept, with clear pathways for future enhancement to a persistent, scalable solution. Environment variables are used for all API keys and sensitive configuration, adhering to best practices for secure application development and deployment across all components.
Project Structure
coding_agent/
├── __init__.py
├── main.py
├── agent_state.py
├── tools.py
├── agent.py
├── config.py
├── requirements.txt
├── Dockerfile
└── README.mdThe coding_agent/ directory serves as the root for our project. The __init__.py file marks it as a Python package, allowing for modular imports. main.py will be the entry point for running our agent, handling initialization and interaction. agent_state.py defines the shared state object that persists across turns in our LangGraph agent, crucial for maintaining context and progress. tools.py encapsulates the external capabilities our agent can use, specifically the E2B code execution sandbox.
The core logic of the LangGraph agent, including node definitions and the graph assembly, resides in agent.py. This separation keeps the orchestration logic clean and manageable. config.py centralizes environment variable loading and other configuration settings, ensuring a clear distinction between code and configuration. requirements.txt lists all Python dependencies, enabling easy environment setup. The Dockerfile provides instructions for containerizing our application, making it portable and deployable. Finally, README.md will contain setup instructions and usage details for the project.
Implementation
Phase 1: Environment Setup and Basic Scaffold
This phase focuses on setting up the project environment, defining the initial LangGraph state, and creating the necessary Python files. We will ensure all dependencies are installed and environment variables are correctly loaded, laying the groundwork for the agent's core logic.
import os
import logging
from typing import TypedDict, List, Annotated
# Configure basic logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# config.py
class Config:
OPENAI_API_KEY: str = os.environ.get("OPENAI_API_KEY", "")
E2B_API_KEY: str = os.environ.get("E2B_API_KEY", "")
def __init__(self):
if not self.OPENAI_API_KEY:
logger.warning("OPENAI_API_KEY not set. Agent may not function correctly.")
if not self.E2B_API_KEY:
logger.warning("E2B_API_KEY not set. Code execution will not work.")
config = Config()
# agent_state.py
class AgentState(TypedDict):
input_problem: str
code: str
output: str
error: str
attempts: int
history: Annotated[List[str], "append"]
solution_found: bool
# requirements.txt
# langgraph>=0.2.0
# langchain-openai>=0.1.7
# e2b>=0.17.0
# python-dotenv==1.0.0
# main.py (initial placeholder)
import logging
logger = logging.getLogger(__name__)
def main():
logger.info("Coding agent scaffold initialized. No operations yet.")
# In later phases, we'll initialize and run the LangGraph agent here.
if __name__ == "__main__":
main()
This phase starts by setting up the project's foundational elements. We begin with
config.py to handle environment variables. Using os.environ.get() is a crucial security practice, preventing sensitive API keys from being hardcoded. We provide default empty strings and log warnings if keys are missing, which is better than crashing immediately, allowing for partial testing or explicit user awareness. A Config class centralizes these settings, making them easily accessible throughout the application.
Next, agent_state.py defines the AgentState using TypedDict. This is the core data structure that LangGraph will use to manage the agent's working memory. TypedDict provides static type checking benefits, making the state structure explicit and reducing runtime errors. Fields like input_problem, code, output, error, and attempts track the agent's progress and current context. The history field uses Annotated[List[str], "append"] to tell LangGraph to append new messages to this list rather than overwriting it, which is useful for maintaining a trace of agent actions or thoughts. solution_found is a boolean flag to indicate successful completion.
We also create a requirements.txt file listing the initial dependencies: langgraph, langchain-openai (for integrating OpenAI models), e2b (for the sandbox), and python-dotenv (useful for local development). Finally, main.py is created as a simple entry point, which will be expanded in subsequent phases to instantiate and run the LangGraph agent. Basic logging is configured globally to provide visibility into the application's startup and any configuration issues, which is vital for debugging and understanding the agent's lifecycle from the very beginning.
Expected outcome
A structured Python project with `config.py`, `agent_state.py`, `requirements.txt`, and an initial `main.py`. Environment variables for OpenAI and E2B are loadable, and basic logging is functional.
How to test
Create a .env file with OPENAI_API_KEY="your_openai_key" and E2B_API_KEY="your_e2b_key". Run pip install -r requirements.txt and then python main.py. You should see the logging messages indicating the scaffold is initialized and warnings if any API keys are missing.
Phase 2: Core Agent Loop with LangGraph
This phase implements the fundamental LangGraph agent, defining the LLM and E2B tool, and structuring the initial graph with nodes for code generation and execution. We will establish the core reasoning and acting loop.
import os
import logging
import json
from typing import TypedDict, List, Annotated, Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from e2b import Sandbox
from tenacity import retry, stop_after_attempt, wait_exponential, before_log
# Configure basic logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# config.py (from Phase 1, ensure it's accessible)
class Config:
OPENAI_API_KEY: str = os.environ.get("OPENAI_API_KEY", "")
E2B_API_KEY: str = os.environ.get("E2B_API_KEY", "")
def __init__(self):
if not self.OPENAI_API_KEY:
logger.error("OPENAI_API_KEY not set. Agent cannot function.")
raise ValueError("OPENAI_API_KEY is required.")
if not self.E2B_API_KEY:
logger.error("E2B_API_KEY not set. Code execution will not work.")
raise ValueError("E2B_API_KEY is required.")
config = Config()
# agent_state.py (from Phase 1, ensure it's accessible)
class AgentState(TypedDict):
input_problem: str
code: str
output: str
error: str
attempts: int
history: Annotated[List[str], "append"]
solution_found: bool
max_attempts: int
# tools.py
class CodeExecutor:
def __init__(self, api_key: str):
self.api_key = api_key
self.sandbox: Sandbox | None = None
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10),
before=before_log(logger, logging.INFO))
def _get_sandbox(self) -> Sandbox:
if self.sandbox is None or not self.sandbox.is_open:
try:
self.sandbox = Sandbox(api_key=self.api_key)
logger.info("E2B Sandbox initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize E2B Sandbox: {e}")
raise
return self.sandbox
def execute_python_code(self, code: str) -> dict:
logger.info("Executing code in E2B Sandbox...")
try:
sandbox = self._get_sandbox()
proc = sandbox.process.start(cmd=f"python -c {json.dumps(code)}", timeout=30)
proc.wait()
stdout = proc.stdout
stderr = proc.stderr
exit_code = proc.exit_code
if exit_code == 0:
logger.info("Code execution successful.")
return {"stdout": stdout, "stderr": stderr, "error": "", "exit_code": exit_code}
else:
error_message = stderr if stderr else f"Process exited with code {exit_code}"
logger.warning(f"Code execution failed: {error_message}")
return {"stdout": stdout, "stderr": stderr, "error": error_message, "exit_code": exit_code}
except Exception as e:
logger.error(f"Error during code execution: {e}")
return {"stdout": "", "stderr": "", "error": str(e), "exit_code": 1}
code_executor = CodeExecutor(api_key=config.E2B_API_KEY)
# agent.py
llm = ChatOpenAI(model="gpt-4o", temperature=0.7, api_key=config.OPENAI_API_KEY)
CODE_GENERATION_PROMPT = ChatPromptTemplate.from_messages([
("system", "You are an expert Python programmer. Your task is to write a Python script to solve the given problem. "
"The code will be executed in a sandbox. Ensure the code is self-contained and prints its final answer to stdout. "
"If libraries are needed, import them at the top. Do not include any explanations, just the code.
Problem: {problem}"),
("user", "Generate Python code for the following problem: {input_problem}")
])
CODE_REFLECTION_PROMPT = ChatPromptTemplate.from_messages([
("system", "You are an expert Python debugger. You are given a problem, the code you wrote, its output, and any error messages. "
"Your task is to analyze the error or incorrect output and provide a corrected version of the Python code. "
"If the output is correct, state 'SOLUTION_FOUND'. Otherwise, provide only the corrected Python code. "
"Do not include any explanations, just the code or 'SOLUTION_FOUND'.
Problem: {problem}
Previous Code: {previous_code}
Execution Output: {execution_output}
Error: {error}
Based on the above, provide the corrected Python code or 'SOLUTION_FOUND'."),
("user", "Refine the code based on the execution results.")
])
# Build the chain for code generation
code_generator = CODE_GENERATION_PROMPT | llm
code_refiner = CODE_REFLECTION_PROMPT | llm
def generate_code(state: AgentState) -> AgentState:
logger.info("Generating initial code...")
problem = state["input_problem"]
response = code_generator.invoke({"problem": problem, "input_problem": problem})
generated_code = response.content
logger.info(f"Generated code: {generated_code[:100]}...")
return {"code": generated_code, "attempts": state["attempts"] + 1, "history": [f"Generated code for attempt {state['attempts'] + 1}. "]}
def execute_code(state: AgentState) -> AgentState:
logger.info("Executing generated code...")
code = state["code"]
execution_result = code_executor.execute_python_code(code)
# Check if a specific success string is present or if there's no error
# For simplicity, we assume no error means potential success for now.
# More robust evaluation will be added in Phase 4.
if not execution_result["error"] and execution_result["exit_code"] == 0:
solution_found = True
output_message = f"Execution successful. Output: {execution_result['stdout']}"
logger.info(output_message)
else:
solution_found = False
output_message = f"Execution failed. Error: {execution_result['error']}. Stdout: {execution_result['stdout']}. Stderr: {execution_result['stderr']}"
logger.warning(output_message)
return {
"output": execution_result["stdout"],
"error": execution_result["error"],
"solution_found": solution_found,
"history": [output_message]
}
def reflect_on_code(state: AgentState) -> AgentState:
logger.info("Reflecting on code execution...")
problem = state["input_problem"]
previous_code = state["code"]
execution_output = state["output"]
error = state["error"]
reflection_response = code_refiner.invoke({
"problem": problem,
"previous_code": previous_code,
"execution_output": execution_output,
"error": error
})
reflection_content = reflection_response.content.strip()
if "SOLUTION_FOUND" in reflection_content:
logger.info("LLM identified solution as successful.")
return {"solution_found": True, "history": ["LLM confirmed solution found."]}
else:
logger.info(f"LLM suggested correction: {reflection_content[:100]}...")
return {"code": reflection_content, "attempts": state["attempts"] + 1, "history": ["LLM provided corrected code."]}
def should_continue(state: AgentState) -> Literal["execute_code", "reflect_on_code", END]:
if state["solution_found"]:
logger.info("Solution found or confirmed by reflection. Ending graph.")
return END
if state["attempts"] >= state["max_attempts"]:
logger.warning(f"Max attempts ({state['max_attempts']}) reached. Ending graph without solution.")
return END
if state["error"] or not state["output"]:
logger.info("Code execution failed or produced no output. Reflecting...")
return "reflect_on_code"
else:
logger.info("Code executed without error, but not yet confirmed. Reflecting for confirmation.")
return "reflect_on_code"
# Build the LangGraph graph
workflow = StateGraph(AgentState)
workflow.add_node("generate_code", generate_code)
workflow.add_node("execute_code", execute_code)
workflow.add_node("reflect_on_code", reflect_on_code)
workflow.set_entry_point("generate_code")
workflow.add_edge("generate_code", "execute_code")
workflow.add_conditional_edges(
"execute_code",
should_continue,
{
"execute_code": "execute_code", # This path should ideally not be hit directly from execute_code, but rather after reflection if needed.
"reflect_on_code": "reflect_on_code",
END: END
},
)
workflow.add_conditional_edges(
"reflect_on_code",
should_continue,
{
"execute_code": "execute_code", # If reflection says 'SOLUTION_FOUND', should_continue will return END. If it gives new code, we re-execute.
"reflect_on_code": "reflect_on_code", # Should not happen, reflection always leads to execute or END
END: END
}
)
app = workflow.compile(checkpointer=MemorySaver())
# main.py (updated)
import logging
from typing import Any
from langgraph.checkpoint.base import Checkpoint
# Assume config, AgentState, app (LangGraph app) are imported or defined as above
logger = logging.getLogger(__name__)
def run_agent(problem: str, max_attempts: int = 5) -> tuple[str, str, Checkpoint | None]:
initial_state = AgentState(
input_problem=problem,
code="",
output="",
error="",
attempts=0,
history=[],
solution_found=False,
max_attempts=max_attempts
)
# Use a specific thread_id for each run for checkpointing if desired
# For this simple example, we'll run it once and get the final state.
# In a real app, you might pass a user ID or session ID as thread_id.
thread_id = "test_run_1"
final_state: AgentState | None = None
last_checkpoint: Checkpoint | None = None
logger.info(f"Starting agent for problem: {problem}")
for s in app.stream(initial_state, config={"configurable": {"thread_id": thread_id}}):
for key, value in s.items():
logger.info(f"Node: {key}, State: {value}")
if key != "__end__": # Store the last non-end state
final_state = value
# Optional: Print history for real-time progress
# if final_state and "history" in final_state and final_state["history"]:
# print(f"Agent Progress: {final_state['history'][-1]}")
if final_state and final_state["solution_found"]:
logger.info(f"Agent successfully solved the problem in {final_state['attempts']} attempts.")
return final_state["code"], final_state["output"], app.get_state(config={"configurable": {"thread_id": thread_id}})
else:
logger.warning("Agent failed to find a solution.")
final_code = final_state["code"] if final_state else "No code generated."
final_output = final_state["output"] if final_state else "No output."
return final_code, final_output, None
if __name__ == "__main__":
# Example problem (can be changed)
problem_statement = "Write a Python function `add_numbers(a, b)` that returns the sum of two numbers. Then call it with 5 and 3 and print the result."
try:
final_code, final_output, checkpoint = run_agent(problem_statement, max_attempts=3)
print("
--- Final Result ---")
print(f"Problem: {problem_statement}")
print(f"Final Code:
{final_code}")
print(f"Final Output:
{final_output}")
if checkpoint:
print(f"Agent State (last checkpoint): {checkpoint.values}")
except ValueError as e:
logger.critical(f"Configuration error: {e}")
except Exception as e:
logger.critical(f"An unexpected error occurred: {e}", exc_info=True)
This phase dramatically expands our agent's capabilities by implementing the core LangGraph workflow. We start by refining
config.py to raise ValueError if critical API keys are missing, ensuring the agent doesn't proceed without essential configurations. The AgentState in agent_state.py is updated to include max_attempts, which will be crucial for controlling the agent's retry loop.
In tools.py, we define CodeExecutor to interface with the E2B sandbox. The _get_sandbox method includes tenacity for robust retry logic with exponential backoff, ensuring that transient network issues or sandbox initialization problems don't immediately fail the agent. This is a production-quality pattern for external service interactions. The execute_python_code method takes a Python code string, starts a process in the E2B sandbox to run it, and captures stdout, stderr, and the exit_code. This detailed feedback is vital for the agent's self-correction. Error handling is explicit, catching Exception during sandbox interaction and returning a structured dictionary with error details.
agent.py is where the LangGraph magic happens. We initialize ChatOpenAI with gpt-4o for powerful language understanding and generation. Two ChatPromptTemplates are defined: CODE_GENERATION_PROMPT guides the LLM to produce Python code for a given problem, and CODE_REFLECTION_PROMPT instructs it to analyze execution results and either correct the code or confirm a solution. These prompts are carefully crafted to elicit specific outputs from the LLM. We then define three core nodes: generate_code, execute_code, and reflect_on_code. Each node takes the AgentState as input and returns a modified AgentState, embodying the functional programming style of LangGraph nodes.
The should_continue function acts as the conditional router for our graph. It determines the next step based on whether a solution has been found, if max attempts are reached, or if reflection is needed. This function is critical for the test-driven loop. Finally, the StateGraph is assembled: nodes are added, an entry point is set, and conditional edges are defined to create the iterative generate-execute-reflect cycle. The MemorySaver is used for checkpointing, allowing the agent's state to be persisted in memory (for this phase), which is invaluable for debugging and potential future restarts. The main.py is updated to instantiate and run this compiled LangGraph application, providing a problem_statement and iterating through the app.stream to observe the agent's progression.
Expected outcome
A functional LangGraph agent that can generate Python code, execute it in an E2B sandbox, and attempt to self-correct based on execution outcomes. The agent will run for a predefined number of attempts or until it finds a solution.
How to test
Ensure your OPENAI_API_KEY and E2B_API_KEY are set in your .env file or environment. Run python main.py. You should see extensive logging output detailing code generation, execution, and reflection steps. The agent should try to solve the add_numbers problem. You might need to adjust the problem statement slightly if GPT-4o perfectly solves it on the first try, to observe the reflection loop.
Phase 3: Test-Driven Self-Correction and Episodic Memory
This phase enhances the agent's self-correction capabilities by explicitly integrating test cases for evaluation and implementing a basic episodic memory to store and retrieve successful solutions. This makes the agent more robust and efficient over time.
import os
import logging
import json
from typing import TypedDict, List, Annotated, Literal, Callable, Dict, Any
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from e2b import Sandbox
from tenacity import retry, stop_after_attempt, wait_exponential, before_log
# Configure basic logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# config.py (from Phase 1, ensure it's accessible)
class Config:
OPENAI_API_KEY: str = os.environ.get("OPENAI_API_KEY", "")
E2B_API_KEY: str = os.environ.get("E2B_API_KEY", "")
def __init__(self):
if not self.OPENAI_API_KEY:
logger.error("OPENAI_API_KEY not set. Agent cannot function.")
raise ValueError("OPENAI_API_KEY is required.")
if not self.E2B_API_KEY:
logger.error("E2B_API_KEY not set. Code execution will not work.")
raise ValueError("E2B_API_KEY is required.")
config = Config()
# agent_state.py (from Phase 1, ensure it's accessible)
class AgentState(TypedDict):
input_problem: str
code: str
output: str
error: str
attempts: int
history: Annotated[List[str], "append"]
solution_found: bool
max_attempts: int
test_cases: List[Dict[str, Any]] # New field for test cases
episodic_memory: Dict[str, str] # New field for episodic memory
# tools.py (updated to include test execution logic)
class CodeExecutor:
def __init__(self, api_key: str):
self.api_key = api_key
self.sandbox: Sandbox | None = None
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10),
before=before_log(logger, logging.INFO))
def _get_sandbox(self) -> Sandbox:
if self.sandbox is None or not self.sandbox.is_open:
try:
self.sandbox = Sandbox(api_key=self.api_key)
logger.info("E2B Sandbox initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize E2B Sandbox: {e}")
raise
return self.sandbox
def execute_python_code(self, code: str) -> dict:
logger.info("Executing code in E2B Sandbox...")
try:
sandbox = self._get_sandbox()
proc = sandbox.process.start(cmd=f"python -c {json.dumps(code)}", timeout=30)
proc.wait()
stdout = proc.stdout
stderr = proc.stderr
exit_code = proc.exit_code
if exit_code == 0:
logger.info("Code execution successful.")
return {"stdout": stdout, "stderr": stderr, "error": "", "exit_code": exit_code}
else:
error_message = stderr if stderr else f"Process exited with code {exit_code}"
logger.warning(f"Code execution failed: {error_message}")
return {"stdout": stdout, "stderr": stderr, "error": error_message, "exit_code": exit_code}
except Exception as e:
logger.error(f"Error during code execution: {e}")
return {"stdout": "", "stderr": "", "error": str(e), "exit_code": 1}
def run_tests(self, code: str, test_cases: List[Dict[str, Any]]) -> Dict[str, Any]:
logger.info("Running tests in E2B Sandbox...")
test_script = code + "
"
test_results = {"passed": 0, "failed": 0, "errors": []}
for i, test_case in enumerate(test_cases):
test_input = test_case.get("input", "")
expected_output = test_case.get("expected_output", "")
# For simplicity, assuming test_input can be directly used in a print statement
# A more robust solution would involve function calls, but this demonstrates the concept.
# A more robust way to run tests: Assume the code defines a function, then call it.
# For example, if problem is "Write add_numbers(a,b)", test_input could be "add_numbers(2,3)"
# and expected_output would be 5.
if "function_name" in test_case and "args" in test_case:
func_call = f"{test_case['function_name']}(*{test_case['args']})"
test_script_with_call = f"{code}
print({func_call})"
else:
# Fallback for simpler cases or direct script execution
test_script_with_call = code # Assume code directly prints output
test_result = self.execute_python_code(test_script_with_call)
if test_result["exit_code"] == 0 and expected_output in test_result["stdout"]:
test_results["passed"] += 1
logger.info(f"Test {i+1} Passed: Input '{test_input}', Expected '{expected_output}', Got '{test_result['stdout'].strip()}'")
else:
test_results["failed"] += 1
error_detail = {
"test_case": test_case,
"stdout": test_result["stdout"],
"stderr": test_result["stderr"],
"error": test_result["error"],
"exit_code": test_result["exit_code"]
}
test_results["errors"].append(error_detail)
logger.warning(f"Test {i+1} Failed: Input '{test_input}', Expected '{expected_output}', Got '{test_result['stdout'].strip()}'. Error: {test_result['error']}")
return test_results
code_executor = CodeExecutor(api_key=config.E2B_API_KEY)
# agent.py (updated to include test evaluation and episodic memory)
llm = ChatOpenAI(model="gpt-4o", temperature=0.7, api_key=config.OPENAI_API_KEY)
CODE_GENERATION_PROMPT = ChatPromptTemplate.from_messages([
("system", "You are an expert Python programmer. Your task is to write a Python script to solve the given problem. "
"The code will be executed in a sandbox. Ensure the code is self-contained and prints its final answer to stdout. "
"If libraries are needed, import them at the top. Do not include any explanations, just the code.
Problem: {problem}
Test Cases: {test_cases_str}"),
("user", "Generate Python code for the following problem: {input_problem}")
])
CODE_REFLECTION_PROMPT = ChatPromptTemplate.from_messages([
("system", "You are an expert Python debugger. You are given a problem, the code you wrote, its output, and any error messages. "
"Your task is to analyze the error or incorrect output and provide a corrected version of the Python code. "
"If the output is correct, state 'SOLUTION_FOUND'. Otherwise, provide only the corrected Python code. "
"Do not include any explanations, just the code or 'SOLUTION_FOUND'.
Problem: {problem}
Previous Code: {previous_code}
Execution Output: {execution_output}
Error: {error}
Test Results: {test_results_str}
Based on the above, provide the corrected Python code or 'SOLUTION_FOUND'."),
("user", "Refine the code based on the execution results.")
])
# Build the chain for code generation
code_generator_chain = CODE_GENERATION_PROMPT | llm
code_refiner_chain = CODE_REFLECTION_PROMPT | llm
EPISODIC_MEMORY: Dict[str, str] = {}
def retrieve_from_memory(problem: str, memory: Dict[str, str]) -> str | None:
# A simple keyword match for demonstration. In a real system, use vector search.
for key, value in memory.items():
if problem.lower() in key.lower() or key.lower() in problem.lower():
logger.info(f"Found similar problem in memory: {key}")
return value
return None
def store_in_memory(problem: str, solution_code: str, memory: Dict[str, str]):
memory[problem] = solution_code
logger.info(f"Stored solution for '{problem}' in episodic memory.")
def generate_code(state: AgentState) -> AgentState:
logger.info("Attempting to retrieve solution from episodic memory...")
problem = state["input_problem"]
episodic_memory = state.get("episodic_memory", {}) # Get memory from state
# Try to retrieve from memory first
recalled_code = retrieve_from_memory(problem, episodic_memory)
if recalled_code:
logger.info("Recalled code from memory, skipping generation.")
return {"code": recalled_code, "attempts": state["attempts"] + 1, "history": ["Recalled solution from episodic memory."]}
logger.info("Generating initial code...")
test_cases_str = json.dumps(state["test_cases"])
response = code_generator_chain.invoke({"problem": problem, "input_problem": problem, "test_cases_str": test_cases_str})
generated_code = response.content
logger.info(f"Generated code: {generated_code[:100]}...")
return {"code": generated_code, "attempts": state["attempts"] + 1, "history": [f"Generated code for attempt {state['attempts'] + 1}. "]}
def execute_code_and_tests(state: AgentState) -> AgentState:
logger.info("Executing generated code and running tests...")
code = state["code"]
test_cases = state["test_cases"]
# First, run the code itself (for general syntax/runtime errors)
initial_execution_result = code_executor.execute_python_code(code)
if initial_execution_result["error"] or initial_execution_result["exit_code"] != 0:
logger.warning("Initial code execution failed, skipping tests.")
return {
"output": initial_execution_result["stdout"],
"error": initial_execution_result["error"],
"solution_found": False,
"history": [f"Initial code execution failed: {initial_execution_result['error']}"]
}
# If initial execution is fine, run the actual tests
test_results = code_executor.run_tests(code, test_cases)
if test_results["failed"] == 0:
solution_found = True
output_message = f"All {test_results['passed']} tests passed. Output: {initial_execution_result['stdout']}"
logger.info(output_message)
# Store successful solution in memory
episodic_memory = state.get("episodic_memory", {})
store_in_memory(state["input_problem"], code, episodic_memory)
return {
"output": initial_execution_result["stdout"],
"error": "",
"solution_found": solution_found,
"history": [output_message],
"episodic_memory": episodic_memory # Update state with modified memory
}
else:
solution_found = False
error_message = f"Tests failed: {test_results['failed']} out of {test_results['passed'] + test_results['failed']} tests failed. Errors: {json.dumps(test_results['errors'])}"
logger.warning(error_message)
return {
"output": initial_execution_result["stdout"],
"error": error_message,
"solution_found": solution_found,
"history": [error_message]
}
def reflect_on_code(state: AgentState) -> AgentState:
logger.info("Reflecting on code execution and test results...")
problem = state["input_problem"]
previous_code = state["code"]
execution_output = state["output"]
error = state["error"]
test_results_str = json.dumps({"passed": 0, "failed": len(state["test_cases"])}) # Placeholder: treat all test cases as failed until actual parsing is implemented
reflection_response = code_refiner_chain.invoke({
"problem": problem,
"previous_code": previous_code,
"execution_output": execution_output,
"error": error,
"test_results_str": test_results_str # Pass test results to reflection
})
reflection_content = reflection_response.content.strip()
if "SOLUTION_FOUND" in reflection_content:
logger.info("LLM identified solution as successful.")
return {"solution_found": True, "history": ["LLM confirmed solution found."]}
else:
logger.info(f"LLM suggested correction: {reflection_content[:100]}...")
return {"code": reflection_content, "attempts": state["attempts"] + 1, "history": ["LLM provided corrected code."]}
def should_continue(state: AgentState) -> Literal["execute_code_and_tests", "reflect_on_code", END]:
if state["solution_found"]:
logger.info("Solution found or confirmed by reflection. Ending graph.")
return END
if state["attempts"] >= state["max_attempts"]:
logger.warning(f"Max attempts ({state['max_attempts']}) reached. Ending graph without solution.")
return END
if state["error"] or not state["output"]:
logger.info("Code execution failed or produced no output. Reflecting...")
return "reflect_on_code"
else:
logger.info("Code executed without error, but not yet confirmed. Reflecting for confirmation.")
return "reflect_on_code"
# Build the LangGraph graph
workflow = StateGraph(AgentState)
workflow.add_node("generate_code", generate_code)
workflow.add_node("execute_code_and_tests", execute_code_and_tests)
workflow.add_node("reflect_on_code", reflect_on_code)
workflow.set_entry_point("generate_code")
workflow.add_edge("generate_code", "execute_code_and_tests")
workflow.add_conditional_edges(
"execute_code_and_tests",
should_continue,
{
"reflect_on_code": "reflect_on_code",
END: END
}
)
workflow.add_conditional_edges(
"reflect_on_code",
should_continue,
{
"execute_code_and_tests": "execute_code_and_tests",
END: END
}
)
app = workflow.compile(checkpointer=MemorySaver())
# main.py (updated to include test cases and episodic memory in initial state)
import logging
from typing import Any
from langgraph.checkpoint.base import Checkpoint
logger = logging.getLogger(__name__)
def run_agent(problem: str, test_cases: List[Dict[str, Any]], max_attempts: int = 5) -> tuple[str, str, Checkpoint | None]:
initial_state = AgentState(
input_problem=problem,
code="",
output="",
error="",
attempts=0,
history=[],
solution_found=False,
max_attempts=max_attempts,
test_cases=test_cases,
episodic_memory={}
)
thread_id = "test_run_1"
final_state: AgentState | None = None
last_checkpoint: Checkpoint | None = None
logger.info(f"Starting agent for problem: {problem}")
for s in app.stream(initial_state, config={"configurable": {"thread_id": thread_id}}):
for key, value in s.items():
logger.info(f"Node: {key}, State: {value}")
if key != "__end__":
final_state = value
if final_state and final_state["solution_found"]:
logger.info(f"Agent successfully solved the problem in {final_state['attempts']} attempts.")
return final_state["code"], final_state["output"], app.get_state(config={"configurable": {"thread_id": thread_id}})
else:
logger.warning("Agent failed to find a solution.")
final_code = final_state["code"] if final_state else "No code generated."
final_output = final_state["output"] if final_state else "No output."
return final_code, final_output, None
if __name__ == "__main__":
# Example problem and test cases
problem_statement = "Write a Python function `multiply_numbers(a, b)` that returns the product of two numbers. Then call it with 4 and 6 and print the result."
test_cases = [
{"function_name": "multiply_numbers", "args": [4, 6], "expected_output": "24"},
{"function_name": "multiply_numbers", "args": [0, 10], "expected_output": "0"},
{"function_name": "multiply_numbers", "args": [-2, 5], "expected_output": "-10"}
]
try:
final_code, final_output, checkpoint = run_agent(problem_statement, test_cases, max_attempts=5)
print("
--- Final Result ---")
print(f"Problem: {problem_statement}")
print(f"Final Code:
{final_code}")
print(f"Final Output:
{final_output}")
if checkpoint:
print(f"Agent State (last checkpoint): {checkpoint.values}")
except ValueError as e:
logger.critical(f"Configuration error: {e}")
except Exception as e:
logger.critical(f"An unexpected error occurred: {e}", exc_info=True)
This phase significantly upgrades the agent's intelligence by introducing explicit test-driven development and a rudimentary episodic memory. In
agent_state.py, we add test_cases (a list of dictionaries defining inputs and expected outputs) and episodic_memory (a dictionary to store successful solutions) to the state. This allows the agent to receive structured test data and to learn from its past successes.
tools.py is enhanced with a run_tests method within the CodeExecutor. This method takes the generated code and a list of test_cases, then executes each test in the E2B sandbox. It constructs a simple test script that calls the function defined by the LLM and captures its output. The results are aggregated, indicating how many tests passed or failed, along with detailed error information for failed tests. This provides concrete, objective feedback to the agent, moving beyond just checking for compilation errors or basic runtime exceptions. The implementation is simplified for demonstration, focusing on the core concept of programmatic evaluation.
In agent.py, the CODE_GENERATION_PROMPT is updated to inform the LLM about the presence of test cases, allowing it to generate code with these in mind. Similarly, CODE_REFLECTION_PROMPT now receives test_results_str to enable more informed debugging. We introduce retrieve_from_memory and store_in_memory functions. retrieve_from_memory performs a simple keyword match (a placeholder for more advanced vector search in a production system) to check if a similar problem has been solved before. store_in_memory saves a problem-solution pair upon successful completion. This episodic memory allows the agent to become more efficient over time by recalling previous solutions.
The generate_code node now first attempts to recall a solution from memory. If found, it bypasses LLM generation, saving tokens and time. The execute_code_and_tests node (renamed from execute_code) now first performs a basic execution to catch immediate syntax/runtime errors, then proceeds to run the test_cases using the run_tests method. If all tests pass, the solution is considered found, and the problem-solution pair is stored in episodic memory. The should_continue function's logic remains similar but now relies on the solution_found flag being set by the test evaluation. The graph edges are updated to reflect the new execute_code_and_tests node. The main.py is updated to pass sample test_cases to the agent, demonstrating its new test-driven capabilities.
Expected outcome
The agent can receive explicit test cases, execute them in the sandbox, and use the test results to guide its self-correction. It will also attempt to retrieve solutions from its in-memory episodic memory before generating new code, and store new successful solutions.
How to test
Ensure your API keys are set. Run python main.py. Observe the logs: the agent should first try to generate code, then execute it and run the provided test cases. If tests fail, it should enter the reflection loop and attempt to correct the code until all tests pass or max attempts are reached. Note: the in-memory episodic_memory dict resets on each run_agent() call. Memory only persists within a single process session; an external store (Redis, PostgreSQL) would be required for cross-run retrieval.
Phase 4: Output Streaming and Production Readiness
This phase focuses on improving the user experience by streaming agent output in real-time and enhancing the production readiness of the agent by introducing more robust logging and error handling for critical components. We'll also refine the main.py to be more production-like.
import os
import logging
import json
import sys
from typing import TypedDict, List, Annotated, Literal, Callable, Dict, Any, Generator
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from e2b import Sandbox
from tenacity import retry, stop_after_attempt, wait_exponential, before_log, retry_if_exception_type
# Configure enhanced logging to stdout for streaming
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', stream=sys.stdout)
logger = logging.getLogger(__name__)
# config.py (from Phase 1, ensure it's accessible)
class Config:
OPENAI_API_KEY: str = os.environ.get("OPENAI_API_KEY", "")
E2B_API_KEY: str = os.environ.get("E2B_API_KEY", "")
def __init__(self):
if not self.OPENAI_API_KEY:
logger.error("OPENAI_API_KEY not set. Agent cannot function.")
raise ValueError("OPENAI_API_KEY is required.")
if not self.E2B_API_KEY:
logger.error("E2B_API_KEY not set. Code execution will not work.")
raise ValueError("E2B_API_KEY is required.")
config = Config()
# agent_state.py (from Phase 1, ensure it's accessible)
class AgentState(TypedDict):
input_problem: str
code: str
output: str
error: str
attempts: int
history: Annotated[List[str], "append"]
solution_found: bool
max_attempts: int
test_cases: List[Dict[str, Any]]
episodic_memory: Dict[str, str]
# tools.py (updated with better error handling for sandbox)
class CodeExecutor:
def __init__(self, api_key: str):
self.api_key = api_key
self.sandbox: Sandbox | None = None
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10),
before=before_log(logger, logging.INFO), retry=retry_if_exception_type(Exception))
def _get_sandbox(self) -> Sandbox:
if self.sandbox is None or not self.sandbox.is_open:
try:
# Ensure sandbox is closed if it somehow became unresponsive
if self.sandbox and self.sandbox.is_open:
self.sandbox.close()
self.sandbox = Sandbox(api_key=self.api_key, timeout=60) # Increased timeout
logger.info("E2B Sandbox initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize E2B Sandbox after retries: {e}")
raise ConnectionError(f"Could not establish E2B sandbox connection: {e}") from e
return self.sandbox
def execute_python_code(self, code: str) -> dict:
logger.info("Executing code in E2B Sandbox...")
try:
sandbox = self._get_sandbox()
proc = sandbox.process.start(cmd=f"python -c {json.dumps(code)}", timeout=60) # Increased timeout
proc.wait()
stdout = proc.stdout
stderr = proc.stderr
exit_code = proc.exit_code
if exit_code == 0:
logger.info("Code execution successful.")
return {"stdout": stdout, "stderr": stderr, "error": "", "exit_code": exit_code}
else:
error_message = stderr if stderr else f"Process exited with code {exit_code}"
logger.warning(f"Code execution failed: {error_message}")
return {"stdout": stdout, "stderr": stderr, "error": error_message, "exit_code": exit_code}
except ConnectionError as e:
logger.error(f"E2B Sandbox connection error during execution: {e}")
return {"stdout": "", "stderr": "", "error": str(e), "exit_code": 1}
except Exception as e:
logger.error(f"Unexpected error during code execution: {e}", exc_info=True)
return {"stdout": "", "stderr": "", "error": str(e), "exit_code": 1}
def run_tests(self, code: str, test_cases: List[Dict[str, Any]]) -> Dict[str, Any]:
logger.info("Running tests in E2B Sandbox...")
test_results = {"passed": 0, "failed": 0, "errors": []}
for i, test_case in enumerate(test_cases):
test_input = test_case.get("input", "")
expected_output = test_case.get("expected_output", "")
if "function_name" in test_case and "args" in test_case:
# Construct a test script that defines the function and then calls it
func_call_args = ", ".join(map(repr, test_case['args'])) # repr handles strings correctly
test_script_with_call = f"{code}
print({test_case['function_name']}({func_call_args}))"
else:
# Fallback for simpler cases or direct script execution (less robust)
test_script_with_call = code
test_result = self.execute_python_code(test_script_with_call)
# Normalize expected_output to string for comparison
expected_output_str = str(expected_output).strip()
actual_stdout = test_result["stdout"].strip()
if test_result["exit_code"] == 0 and expected_output_str == actual_stdout:
test_results["passed"] += 1
logger.info(f"Test {i+1} Passed: Input '{test_input}', Expected '{expected_output_str}', Got '{actual_stdout}'")
else:
test_results["failed"] += 1
error_detail = {
"test_case": test_case,
"stdout": actual_stdout,
"stderr": test_result["stderr"],
"error": test_result["error"],
"exit_code": test_result["exit_code"]
}
test_results["errors"].append(error_detail)
logger.warning(f"Test {i+1} Failed: Input '{test_input}', Expected '{expected_output_str}', Got '{actual_stdout}'. Error: {test_result['error']}")
return test_results
code_executor = CodeExecutor(api_key=config.E2B_API_KEY)
# agent.py (minor prompt adjustments for clarity, same core logic)
llm = ChatOpenAI(model="gpt-4o", temperature=0.7, api_key=config.OPENAI_API_KEY, streaming=True)
CODE_GENERATION_PROMPT = ChatPromptTemplate.from_messages([
("system", "You are an expert Python programmer. Your task is to write a Python script to solve the given problem. "
"The code will be executed in a sandbox. Ensure the code is self-contained and prints its final answer to stdout. "
"If libraries are needed, import them at the top. Do not include any explanations, just the code. "
"You will be provided with test cases to validate your code. Your function signature must match the problem description if a function is requested.
Problem: {problem}
Test Cases: {test_cases_str}"),
("user", "Generate Python code for the following problem: {input_problem}")
])
CODE_REFLECTION_PROMPT = ChatPromptTemplate.from_messages([
("system", "You are an expert Python debugger. You are given a problem, the code you wrote, its execution output, errors, and test results. "
"Your task is to analyze the failures and provide a corrected version of the Python code. "
"If the code passes all tests and is correct, output 'SOLUTION_FOUND'. Otherwise, output only the corrected Python code. "
"Do not include any explanations, comments, or extra text, just the code or 'SOLUTION_FOUND'.
Problem: {problem}
Previous Code: {previous_code}
Execution Output: {execution_output}
Error: {error}
Test Results: {test_results_str}
Based on the above, provide the corrected Python code or 'SOLUTION_FOUND'."),
("user", "Refine the code based on the execution results.")
])
# Build the chain for code generation
code_generator_chain = CODE_GENERATION_PROMPT | llm
code_refiner_chain = CODE_REFLECTION_PROMPT | llm
# EPISODIC_MEMORY is now part of AgentState, no global variable needed here.
def retrieve_from_memory(problem: str, memory: Dict[str, str]) -> str | None:
for key, value in memory.items():
if problem.lower() in key.lower() or key.lower() in problem.lower():
logger.info(f"Found similar problem in memory: {key}")
return value
return None
def store_in_memory(problem: str, solution_code: str, memory: Dict[str, str]):
memory[problem] = solution_code
logger.info(f"Stored solution for '{problem}' in episodic memory.")
def generate_code(state: AgentState) -> AgentState:
logger.info("Attempting to retrieve solution from episodic memory...")
problem = state["input_problem"]
episodic_memory = state.get("episodic_memory", {})
recalled_code = retrieve_from_memory(problem, episodic_memory)
if recalled_code:
logger.info("Recalled code from memory, skipping generation.")
return {"code": recalled_code, "attempts": state["attempts"] + 1, "history": ["Recalled solution from episodic memory."]}
logger.info("Generating initial code...")
test_cases_str = json.dumps(state["test_cases"])
response = code_generator_chain.invoke({"problem": problem, "input_problem": problem, "test_cases_str": test_cases_str})
generated_code = response.content
logger.info(f"Generated code: {generated_code[:100]}...")
return {"code": generated_code, "attempts": state["attempts"] + 1, "history": [f"Generated code for attempt {state['attempts'] + 1}. "]}
def execute_code_and_tests(state: AgentState) -> AgentState:
logger.info("Executing generated code and running tests...")
code = state["code"]
test_cases = state["test_cases"]
initial_execution_result = code_executor.execute_python_code(code)
if initial_execution_result["error"] or initial_execution_result["exit_code"] != 0:
logger.warning("Initial code execution failed, skipping tests.")
return {
"output": initial_execution_result["stdout"],
"error": initial_execution_result["error"],
"solution_found": False,
"history": [f"Initial code execution failed: {initial_execution_result['error']}"]
}
test_results = code_executor.run_tests(code, test_cases)
if test_results["failed"] == 0:
solution_found = True
output_message = f"All {test_results['passed']} tests passed. Output: {initial_execution_result['stdout']}"
logger.info(output_message)
episodic_memory = state.get("episodic_memory", {})
store_in_memory(state["input_problem"], code, episodic_memory)
return {
"output": initial_execution_result["stdout"],
"error": "",
"solution_found": solution_found,
"history": [output_message],
"episodic_memory": episodic_memory
}
else:
solution_found = False
error_message = f"Tests failed: {test_results['failed']} out of {test_results['passed'] + test_results['failed']} tests failed. Errors: {json.dumps(test_results['errors'])}"
logger.warning(error_message)
return {
"output": initial_execution_result["stdout"],
"error": error_message,
"solution_found": solution_found,
"history": [error_message]
}
def reflect_on_code(state: AgentState) -> AgentState:
logger.info("Reflecting on code execution and test results...")
problem = state["input_problem"]
previous_code = state["code"]
execution_output = state["output"]
error = state["error"]
test_results_summary = {"passed": 0, "failed": 0}
if state["error"] and "Tests failed" in state["error"]:
try:
# Extract actual test results from error message if available
error_json_str = state["error"].split("Errors: ", 1)[1]
error_details = json.loads(error_json_str)
test_results_summary["failed"] = len(error_details)
test_results_summary["passed"] = len(state["test_cases"]) - test_results_summary["failed"]
except (IndexError, json.JSONDecodeError):
logger.warning("Could not parse detailed test results from error message.")
test_results_summary["failed"] = 1 # Indicate at least one failure
else:
# If no explicit test failure error, assume 0 failed for reflection purposes
test_results_summary["passed"] = len(state["test_cases"])
reflection_response = code_refiner_chain.invoke({
"problem": problem,
"previous_code": previous_code,
"execution_output": execution_output,
"error": error,
"test_results_str": json.dumps(test_results_summary)
})
reflection_content = reflection_response.content.strip()
if "SOLUTION_FOUND" in reflection_content:
logger.info("LLM identified solution as successful.")
return {"solution_found": True, "history": ["LLM confirmed solution found."]}
else:
logger.info(f"LLM suggested correction: {reflection_content[:100]}...")
return {"code": reflection_content, "attempts": state["attempts"] + 1, "history": ["LLM provided corrected code."]}
def should_continue(state: AgentState) -> Literal["execute_code_and_tests", "reflect_on_code", END]:
if state["solution_found"]:
logger.info("Solution found or confirmed by reflection. Ending graph.")
return END
if state["attempts"] >= state["max_attempts"]:
logger.warning(f"Max attempts ({state['max_attempts']}) reached. Ending graph without solution.")
return END
# Continue to reflect if there was an error, or if tests failed (indicated by error message)
# or if no solution was found yet but no explicit error (means LLM needs to confirm)
if state["error"] or not state["solution_found"]:
logger.info("Code execution failed or solution not yet confirmed. Reflecting...")
return "reflect_on_code"
else: # This path should logically not be hit if solution_found is false and there's no error
logger.warning("Unexpected state in should_continue. Forcing reflection.")
return "reflect_on_code"
# Build the LangGraph graph
workflow = StateGraph(AgentState)
def should_stop_or_reexecute(state: AgentState) -> Literal["execute_code_and_tests", END]:
"""Routing from reflect_on_code: return to execution unless done."""
if state["solution_found"] or state["attempts"] >= state["max_attempts"]:
return END
return "execute_code_and_tests"
workflow.add_node("generate_code", generate_code)
workflow.add_node("execute_code_and_tests", execute_code_and_tests)
workflow.add_node("reflect_on_code", reflect_on_code)
workflow.set_entry_point("generate_code")
workflow.add_edge("generate_code", "execute_code_and_tests")
workflow.add_conditional_edges(
"execute_code_and_tests",
should_continue,
{
"reflect_on_code": "reflect_on_code",
END: END
}
)
workflow.add_conditional_edges(
"reflect_on_code",
should_stop_or_reexecute,
{
"execute_code_and_tests": "execute_code_and_tests",
END: END
}
)
app = workflow.compile(checkpointer=MemorySaver())
# main.py (updated for streaming and more robust execution)
import logging
import sys
from typing import Any, Generator
from langgraph.checkpoint.base import Checkpoint
logger = logging.getLogger(__name__)
def stream_agent_output(problem: str, test_cases: List[Dict[str, Any]], max_attempts: int = 5) -> Generator[Dict[str, Any], None, None]:
initial_state = AgentState(
input_problem=problem,
code="",
output="",
error="",
attempts=0,
history=[],
solution_found=False,
max_attempts=max_attempts,
test_cases=test_cases,
episodic_memory={}
)
thread_id = "streaming_run_1"
yield {"event": "start", "problem": problem, "max_attempts": max_attempts}
current_state: AgentState | None = None
try:
for s in app.stream(initial_state, config={"configurable": {"thread_id": thread_id}}):
for key, value in s.items():
if key == "__end__":
yield {"event": "end_of_stream", "final_state": current_state}
return # Exit generator after final state
current_state = value # Keep track of the latest state
yield {"event": "node_update", "node": key, "state": value}
# Stream specific parts of history for user feedback
if "history" in value and value["history"]:
yield {"event": "progress", "message": value["history"][-1]}
if "code" in value and key == "generate_code":
yield {"event": "code_generated", "code": value["code"]}
if "output" in value and key == "execute_code_and_tests":
yield {"event": "execution_result", "output": value["output"], "error": value["error"], "solution_found": value["solution_found"]}
if "code" in value and key == "reflect_on_code" and not value["solution_found"]:
yield {"event": "code_corrected", "code": value["code"]}
# If loop finishes without hitting __end__ (e.g., due to an error before END node)
if current_state and current_state["solution_found"]:
yield {"event": "solution_found", "code": current_state["code"], "output": current_state["output"]}
else:
yield {"event": "failure", "message": "Agent failed to find a solution.", "final_state": current_state}
except ValueError as e:
logger.critical(f"Configuration error during streaming: {e}")
yield {"event": "error", "message": f"Configuration error: {e}"}
except Exception as e:
logger.critical(f"An unexpected error occurred during streaming: {e}", exc_info=True)
yield {"event": "error", "message": f"Unexpected error: {str(e)}"}
if __name__ == "__main__":
problem_statement = "Write a Python function `is_prime(n)` that returns True if n is a prime number, False otherwise. Then call it with 7 and 10 and print the results."
test_cases = [
{"function_name": "is_prime", "args": [7], "expected_output": "True"},
{"function_name": "is_prime", "args": [10], "expected_output": "False"},
{"function_name": "is_prime", "args": [2], "expected_output": "True"},
{"function_name": "is_prime", "args": [1], "expected_output": "False"}
]
print("
--- Starting Agent Stream ---")
for event in stream_agent_output(problem_statement, test_cases, max_attempts=5):
if event["event"] == "progress":
print(f"[AGENT PROGRESS]: {event['message']}")
elif event["event"] == "code_generated":
print(f"[CODE GENERATED]:
{event['code']}")
elif event["event"] == "execution_result":
print(f"[EXECUTION RESULT]: Solution found: {event['solution_found']}, Error: {event['error'] or 'None'}
Output: {event['output']}")
elif event["event"] == "code_corrected":
print(f"[CODE CORRECTED]:
{event['code']}")
elif event["event"] == "solution_found":
print("
--- FINAL SOLUTION ---")
print(f"Code:
{event['code']}")
print(f"Output:
{event['output']}")
elif event["event"] == "failure":
print("
--- AGENT FAILED ---")
print(f"Message: {event['message']}")
print(f"Final State: {event['final_state']}")
elif event["event"] == "error":
print(f"[CRITICAL ERROR]: {event['message']}")
elif event["event"] == "end_of_stream":
print("--- End of Agent Stream ---")
In this final implementation phase, we focus on making the agent more user-friendly through output streaming and more robust for production environments. First,
logging is configured to output to sys.stdout, which is standard for containerized applications and facilitates easier log aggregation and streaming. The ChatOpenAI instance is initialized with streaming=True to allow for real-time token generation, although for code generation, the full response is often needed before execution.
tools.py receives further hardening in CodeExecutor. The _get_sandbox method's retry decorator is updated to specifically retry Exception types, ensuring that transient issues with E2B sandbox initialization are handled. We also add logic to explicitly close an unresponsive sandbox before re-initializing it and increase the sandbox and process timeouts to 60 seconds, which is crucial for complex code execution that might take longer. More robust error handling for ConnectionError and generic Exception is added within execute_python_code to catch and log potential issues during the execution phase, providing more informative error messages.
The run_tests method in tools.py is improved to construct more accurate test scripts, especially for functions. It uses repr() for arguments to handle various data types correctly, making the test execution more reliable. The comparison for expected_output is also made more robust by stripping whitespace and converting to string, preventing subtle mismatches. The reflect_on_code node in agent.py is updated to better parse detailed test results from the error message if available, providing more granular feedback to the LLM for correction.
The most significant change is in main.py, which is refactored into a stream_agent_output generator function. This function iterates through the app.stream() output from LangGraph, yielding structured events (start, node_update, progress, code_generated, execution_result, code_corrected, solution_found, failure, error, end_of_stream). This pattern is essential for building interactive UIs or integrating with external systems that require real-time updates on the agent's progress. Instead of just printing the final state, we now get a detailed play-by-play of the agent's internal thought process and actions, which is invaluable for debugging, monitoring, and providing a rich user experience. Error handling is also integrated into the streaming function to catch and yield error events gracefully.
Expected outcome
The agent now streams its progress and significant state changes in real-time. The sandbox integration is more resilient to transient errors, and the overall system is more robust, providing clearer feedback during its operation.
How to test
Run python main.py again. You should now see a continuous stream of events, including when code is generated, executed, corrected, and when tests pass or fail. Observe how the agent handles the is_prime problem, including any corrections it makes. Verify that the SOLUTION_FOUND event is emitted upon success.
Testing & Evaluation
Testing and evaluation for a coding agent requires a multi-faceted approach, balancing functional correctness with the nuanced behavior of LLMs. Our primary goal is to verify that the agent can reliably generate, execute, and self-correct code to solve a given problem.
Functional Testing
Functional testing is paramount. For each problem, we define a set of concrete test cases, as demonstrated in our test_cases structure. These test cases are executed within the E2B sandbox, providing objective pass/fail criteria. This includes:
- Unit Test Coverage: For problems requiring a function, ensure the agent's generated function handles various inputs, including edge cases (e.g.,
is_prime(1),multiply_numbers(0, 10)). - Output Format Validation: Verify that the agent's final output (e.g., printed result) matches the expected format.
- Error Handling: Inject scenarios where the LLM might generate syntactically incorrect Python or code that causes runtime errors (e.g., division by zero). The agent should enter its reflection loop and attempt to correct these.
An automated evaluation script would iterate through a predefined suite of problems, each with its own test_cases. For each problem, it would invoke the stream_agent_output function, collect the final state, and assert on solution_found being True and the final code passing all test_cases.
Automated Evaluation Example
Here's an example of an automated test runner script using pytest for a problem test_coding_agent.py:
import pytest
import logging
from typing import Dict, Any, List
from main import stream_agent_output, AgentState # Assuming main.py is in the same directory
logger = logging.getLogger(__name__)
logger.setLevel(logging.CRITICAL) # Suppress agent's internal INFO logs during tests
# Define test problems and their expected outcomes
TEST_PROBLEMS = [
{
"problem": "Write a Python function `power(base, exp)` that calculates base raised to the power of exp. Assume exp is a non-negative integer. Then call it with (2, 3) and (5, 0) and print the results.",
"test_cases": [
{"function_name": "power", "args": [2, 3], "expected_output": "8"},
{"function_name": "power", "args": [5, 0], "expected_output": "1"},
{"function_name": "power", "args": [10, 1], "expected_output": "10"}
],
"expected_solution": True
},
{
"problem": "Write a Python function `reverse_string(s)` that returns the input string reversed. Then call it with 'hello' and 'python' and print the results.",
"test_cases": [
{"function_name": "reverse_string", "args": ["hello"], "expected_output": "olleh"},
{"function_name": "reverse_string", "args": ["python"], "expected_output": "nohtyp"},
{"function_name": "reverse_string", "args": ["a"], "expected_output": "a"}
],
"expected_solution": True
}
]
@pytest.mark.parametrize("test_data", TEST_PROBLEMS)
def test_agent_solves_problem(test_data: Dict[str, Any]):
problem = test_data["problem"]
test_cases = test_data["test_cases"]
expected_solution = test_data["expected_solution"]
max_attempts = 5
logger.info(f"Running test for problem: {problem}")
final_code = ""
final_output = ""
solution_found = False
final_state: AgentState | None = None
for event in stream_agent_output(problem, test_cases, max_attempts):
if event["event"] == "solution_found":
final_code = event["code"]
final_output = event["output"]
solution_found = True
if "final_state" in event:
final_state = event["final_state"]
assert solution_found == expected_solution, f"Agent did not {'find' if expected_solution else 'fail to find'} solution for problem: {problem}"
if solution_found:
# Re-run the tests on the final code to confirm correctness independently
# This requires an instance of CodeExecutor to be available for testing
from tools import CodeExecutor # Assuming tools.py is accessible
from config import config
test_executor = CodeExecutor(api_key=config.E2B_API_KEY)
validation_results = test_executor.run_tests(final_code, test_cases)
assert validation_results["failed"] == 0, f"Final code failed validation tests for problem: {problem}. Errors: {validation_results['errors']}"
logger.info(f"Successfully validated solution for problem: {problem}")
else:
logger.warning(f"Agent failed to find a solution for problem: {problem}. Final state: {final_state}")
# To run these tests:
# 1. Ensure pytest, e2b, langchain-openai, langgraph are installed.
# 2. Set OPENAI_API_KEY and E2B_API_KEY environment variables.
# 3. Navigate to the project root and run: pytest test_coding_agent.pyLLM-as-a-Judge Evaluation
While functional tests are objective, the *quality* of the generated code (e.g., efficiency, readability, adherence to best practices) can be evaluated using an LLM-as-a-Judge. This involves a separate LLM prompted to critique the agent's solution based on criteria beyond simple correctness. This is particularly useful for assessing solutions that pass tests but are suboptimal.
Failure Scenario Testing
- Sandbox Errors: Test with code that tries to access restricted resources or exceeds execution time limits. The agent should gracefully handle E2B errors and reflect appropriately.
- LLM Hallucinations: Provide prompts that are ambiguous or require highly specialized knowledge, leading the LLM to generate incorrect but syntactically valid code. The reflection loop should catch these via test failures.
- API Rate Limits/Downtime: Simulate API failures for OpenAI or E2B to ensure retry mechanisms (like
tenacity) and fallback strategies are effective.
Edge Case Coverage
Ensure test cases cover typical edge cases for coding problems, such as empty inputs, zero values, negative numbers, or maximum/minimum constraints. This helps the agent learn to handle a broader range of scenarios.
Performance Validation
Measure the end-to-end latency of solving problems, especially those requiring multiple reflection steps. Monitor token usage to understand cost implications. Tools like OpenTelemetry can be integrated to trace agent execution paths and identify bottlenecks. For this project, a simple time measurement in main.py can give initial insights.
Regression Testing
As the agent's capabilities evolve, maintain a growing suite of solved problems in the TEST_PROBLEMS array. Regularly running this suite ensures that new features or prompt changes do not inadvertently break previously working solutions. This also validates the effectiveness of the episodic memory if a problem is solved faster on subsequent runs.
Deployment
Deploying the coding agent involves containerizing the application and orchestrating it in a production environment like Kubernetes or a cloud-native platform. This ensures reproducibility, scalability, and robust operation.
1. Containerization with Docker
First, we create a Dockerfile to package our Python application and its dependencies into a Docker image. This image will contain everything needed to run the agent.
Dockerfile
# Use an official Python runtime as a parent image
FROM python:3.10-slim-buster
# Set the working directory in the container
WORKDIR /app
# Install system dependencies required by e2b or other libraries
# For example, git might be needed for some e2b operations or specific packages.
# apt-get update && apt-get install -y --no-install-recommends git && rm -rf /var/lib/apt/lists/*
# Copy the requirements file into the container
COPY requirements.txt .
# Install any needed packages specified in requirements.txt
RUN pip install --no-cache-dir -r requirements.txt
# Copy the entire project into the container
COPY . .
# Expose port 8000 if you plan to add a web interface later
# EXPOSE 8000
# Define environment variables for API keys (never hardcode them here)
ENV OPENAI_API_KEY=""
ENV E2B_API_KEY=""
# Run the application (replace with your actual entry point if it's a web server)
# For a simple agent, we might just run a script or keep it idle for API calls.
# CMD ["python", "main.py"]
# For a more robust entrypoint, consider a wrapper script or a web server.
# For now, we'll keep it as a simple script that can be invoked.
CMD ["/bin/bash"]Build and Run Locally
docker build -t coding-agent .
docker run -it --rm \
-e OPENAI_API_KEY="your_openai_key" \
-e E2B_API_KEY="your_e2b_key" \
coding-agent \
python main.py2. Kubernetes Deployment
For production, Kubernetes provides orchestration, scaling, and self-healing. We'll define a Deployment and a Secret.
k8s/agent-secret.yaml
apiVersion: v1
kind: Secret
metadata:
name: coding-agent-secrets
type: Opaque
data:
OPENAI_API_KEY: <base64_encoded_openai_key>
E2B_API_KEY: <base64_encoded_e2b_key>Replace <base64_encoded_openai_key> and <base64_encoded_e2b_key> with base64 encoded versions of your actual keys (echo -n 'your_key' | base64).
k8s/agent-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: coding-agent-deployment
labels:
app: coding-agent
spec:
replicas: 1
selector:
matchLabels:
app: coding-agent
template:
metadata:
labels:
app: coding-agent
spec:
containers:
- name: coding-agent
image: your_docker_registry/coding-agent:latest # Replace with your image
command: ["python", "main.py"]
envFrom:
- secretRef:
name: coding-agent-secrets
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "1Gi"
cpu: "1000m"
readinessProbe:
exec:
command: ["python", "-c", "import os; exit(0) if os.environ.get('OPENAI_API_KEY') and os.environ.get('E2B_API_KEY') else exit(1)"]
initialDelaySeconds: 5
periodSeconds: 10
livenessProbe:
exec:
command: ["python", "-c", "import logging; logging.basicConfig(level=logging.INFO); logging.info('Agent is alive'); exit(0)"]
initialDelaySeconds: 15
periodSeconds: 20Deployment Steps:
- Push your Docker image to a registry:
docker tag coding-agent your_docker_registry/coding-agent:latest && docker push your_docker_registry/coding-agent:latest - Apply the secret:
kubectl apply -f k8s/agent-secret.yaml - Apply the deployment:
kubectl apply -f k8s/agent-deployment.yaml
3. Secrets Management
As shown, Kubernetes Secrets are used to inject API keys as environment variables into the container. Never embed secrets directly in your Dockerfile or deployment manifests. For highly sensitive data or advanced scenarios, consider external secret management solutions like HashiCorp Vault or cloud provider KMS services.
4. Health Check and Readiness Probe
- Readiness Probe: The
readinessProbechecks if the container is ready to serve traffic. Our probe ensures that the necessary environment variables (OPENAI_API_KEY,E2B_API_KEY) are present, which are critical for the agent's functionality. If these are missing, the pod won't be marked as ready. - Liveness Probe: The
livenessProbeverifies that the application is still running and healthy. A simple Python script that logs a message and exits successfully can indicate the Python runtime is active. For a more sophisticated agent (e.g., one with a web server), this would involve an HTTP endpoint check.
5. Scaling Considerations
- Horizontal Pod Autoscaler (HPA): Configure an HPA to scale the number of agent replicas based on CPU utilization or custom metrics (e.g., number of pending tasks in a queue). This ensures the agent can handle varying workloads.
- E2B Sandbox Limits: Be mindful of E2B API rate limits and concurrent sandbox limits. If scaling up, ensure your E2B plan supports the increased usage. Each agent replica will likely consume its own sandbox instances.
- LLM Rate Limits: OpenAI and other LLM providers have rate limits. Implement exponential backoff and potentially integrate with a token bucket algorithm at a global level if multiple agents share the same API key, or use separate API keys per instance if possible.
6. Rollback Strategy
Kubernetes inherently supports rollbacks. If a new deployment introduces issues:
- Monitor: Closely monitor logs and metrics after a new deployment.
- Rollback Command: If issues arise, use
kubectl rollout undo deployment/coding-agent-deploymentto revert to the previous stable version of the deployment. Ensure your Docker image tags are immutable (e.g.,v1.0.1instead oflatest) to guarantee you're rolling back to a specific, known-good image. - Automated Rollbacks: Integrate CI/CD pipelines with automated rollback triggers based on failing health checks or critical error alerts, as discussed in /agents/production-engineering/automated-rollbacks-for-agents/.
Production Hardening
Moving the coding agent from development to production requires diligent attention to various hardening aspects to ensure its security, reliability, and cost-effectiveness.
Security
- Sandboxing Unexpected Code Execution: The use of E2B is critical for isolating LLM-generated code. Ensure E2B's security features (e.g., network isolation, resource limits) are fully leveraged. Regularly review E2B's security posture and updates. This directly addresses OWASP ASI05: Sandboxing Unexpected Code Execution.
- Input Validation and Sanitization: Although the agent generates its own code, user inputs (problem statements, test cases) should be validated to prevent prompt injection or malicious data from influencing code generation or reflection. While LLMs are robust, sanitizing inputs reduces attack surface.
- Secrets Management: Never hardcode API keys. Use Kubernetes Secrets (as shown in deployment) or more advanced solutions like HashiCorp Vault or cloud-specific secret managers (AWS Secrets Manager, Azure Key Vault). Implement RBAC for secrets access.
- Principle of Least Privilege: Ensure the container running the agent has only the necessary permissions. Avoid running as root. Limit network access to only external services it needs (OpenAI, E2B).
- Dependency Scanning: Use tools like Trivy or Snyk in your CI/CD pipeline to scan Docker images and
requirements.txtfor known vulnerabilities in third-party libraries.
Observability
- Structured Logging: Implement structured logging (e.g., JSON logs) for all agent activities, including node transitions, tool calls, LLM inputs/outputs, and error messages. This makes logs easier to parse and analyze with tools like Elastic Stack or Splunk.
- Tracing Multi-Agent Workflows: Integrate OpenTelemetry for end-to-end tracing of the agent's execution. This is crucial for understanding complex LangGraph cycles, identifying bottlenecks, and debugging non-deterministic behavior. Define custom spans for LLM calls and sandbox executions. See /agents/protocols/tracing-multi-agent-workflows/.
- Metrics and Alerts: Collect metrics on agent performance (e.g., latency per problem, success rate, number of reflection steps, token usage). Set up alerts for critical failures, high error rates, or performance degradation.
- LLM Observability: Monitor LLM-specific metrics like token usage, cost per run, and model response times. This helps in understanding the economic impact and performance of the chosen LLM.
Reliability
- Retry Logic and Exponential Backoff: Implement
tenacityfor all external API calls (OpenAI, E2B) to handle transient failures gracefully. Ensure backoff strategies prevent overwhelming upstream services. - Fallback LLM Providers: Consider configuring a fallback LLM provider in case the primary one experiences outages or rate limit exhaustion. This can be integrated into the
ChatOpenAIwrapper. See /agents/protocols/fallback-llm-provider-reliability/. - Idempotency: Design agent operations to be idempotent where possible. If a message is processed twice (e.g., due to retry), it should produce the same result or not cause unintended side effects.
- State Persistence (Checkpointers): While
MemorySaveris used in this guide, for production, replace it with a persistent checkpointer likePostgreSQLSaveror Redis. This allows the agent to resume from its last known state after restarts or failures, preventing loss of progress. See /agents/frameworks/langgraph-checkpointers-state-persistence/.
Rate Limiting
- API Rate Limiting: Implement client-side rate limiting for LLM and E2B API calls to stay within provider limits. This can be done via libraries or by leveraging Kubernetes service meshes.
- Concurrency Control: Limit the number of concurrent sandboxes or LLM calls per agent instance to manage resource consumption and avoid hitting external API limits.
Authentication
- Service-to-Service Authentication: If the agent interacts with other internal services, use secure authentication mechanisms (e.g., OAuth, mTLS) rather than API keys passed directly.
Governance
- Prompt Versioning: Maintain versions of your
CODE_GENERATION_PROMPTandCODE_REFLECTION_PROMPT. Changes to prompts can significantly alter agent behavior and should be tracked and tested. See /agents/production-engineering/prompt-versioning-drift-detection/. - Human-in-the-Loop (HITL): For critical or complex problems, consider an interrupt node in LangGraph to allow human review and approval of generated code before execution, or to assist in debugging. See /agents/production-engineering/hitl-escalation-state-machines/.
Cost Optimisation
- Token Monitoring: Continuously monitor token usage for LLM calls. Optimize prompts to be concise yet effective to reduce costs.
- Episodic Memory Effectiveness: Evaluate the hit rate of episodic memory. A high hit rate means fewer LLM calls, directly impacting cost savings. Refine memory retrieval strategies to maximize this.
- Resource Allocation: Fine-tune Kubernetes resource requests and limits (CPU, memory) to match the agent's actual needs, avoiding over-provisioning.
Compliance
- Data Residency: If handling sensitive data, ensure the E2B sandbox and LLM provider adhere to data residency requirements. Understand where code is executed and data is processed. See /agents/production-engineering/data-residency-context-governance/.
- Logging and Audit Trails: Ensure all significant agent actions and decisions are logged for auditability, especially in regulated environments. This is crucial for demonstrating compliance and investigating incidents.
Learning Objectives
- By the end of this project, you will be able to design and implement a stateful agent workflow using LangGraph's state machine capabilities.
- By the end of this project, you will be able to integrate and utilize external code execution environments like E2B for secure sandboxed tool use.
- By the end of this project, you will be able to implement agent reflection and self-correction mechanisms to improve problem-solving accuracy.
- By the end of this project, you will be able to incorporate episodic memory to store and retrieve past successful solutions, enhancing agent efficiency.
- By the end of this project, you will be able to implement robust error handling and retry logic within an agentic workflow to ensure reliability.
- By the end of this project, you will be able to containerize a LangGraph agent for deployment and understand basic production hardening considerations.
Conceptual Prerequisites
What Is Agentic AI?
A foundational understanding of what AI agents are, their components, and how they differ from traditional LLM applications is crucial for grasping the purpose and mechanics of this project.
State Machines for AI Agents
Familiarity with state machine concepts is essential, as LangGraph is built around this paradigm. Understanding states, nodes, and edges will make the graph construction much clearer.
Agent Reflection and Self-Correction
Understanding how agents can analyze their own outputs and failures to improve is key to appreciating the self-correction loop implemented in this coding agent.
Episodic Memory: Vector Databases for Agent History
A basic understanding of episodic memory and how it can be used to store and retrieve past experiences will help in understanding how the agent learns from successful solutions.
Technical Prerequisites
Python 3.9+
All code examples are written in Python. Ensure you have a recent version installed and are comfortable with its syntax and package management (pip).
OpenAI API Key
The project uses OpenAI's GPT models. You will need an OpenAI API key with access to models like `gpt-4o` or `gpt-4o-mini` configured as an environment variable.
E2B API Key
The E2B sandbox is used for secure code execution. You will need an E2B API key, which can be obtained from their website, and configured as an environment variable.
Docker
For deployment, you will need Docker installed to build and run the containerized application.