Building Autonomous Agent Loops With LangChain and OpenAI Function Calling A Practical Tutorial

# Create a virtual environment (recommended)
python -m venv venv
source venv/bin/activate   # On Windows: venv\Scripts\activate

# Install dependencies
pip install langchain openai pydantic python-dotenv tqdm

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from typing import List, Dict, Any

class SearchArgs(BaseModel):
    query: str = Field(..., description="The search string the user wants to look up.")

class APICallArgs(BaseModel):
    endpoint: str = Field(..., description="Full URL of the external API.")
    payload: Dict[str, Any] = Field(..., description="JSON payload to POST to the endpoint.")

class FormatArgs(BaseModel):
    data: List[Dict[str, Any]] = Field(..., description="List of dictionaries returned from a previous step.")

import random
import json
import requests
from time import sleep

def search_documents(query: str) -> List[Dict]:
    """
    Mocked vector store search. In production replace with Pinecone, Weaviate, etc.
    """
    # Simulated latency
    sleep(0.5)
    # Return 3 dummy results
    results = [
        {"title": f"Result {i} for {query}", "snippet": f"This is a short excerpt about {query} #{i}"}
        for i in range(1, 4)
    ]
    return results

def call_external_api(endpoint: str, payload: Dict) -> Dict:
    """
    Simple wrapper around a POST request. Handles errors & returns JSON.
    """
    try:
        response = requests.post(endpoint, json=payload, timeout=5)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        # In a real agent you might want to surface the error to the LLM
        return {"error": str(e), "status_code": getattr(e.response, "status_code", None)}

def format_response(data: List[Dict]) -> str:
    """
    Turns a list of result dicts into a concise paragraph.
    """
    if not data:
        return "I couldn't find any relevant information."
    lines = [f"- **{item['title']}**: {item['snippet']}" for item in data]
    return "Here are the top results:\n" + "\n".join(lines)

from langchain.tools import Tool
from langchain.utilities import OpenAIFunctions

# Wrap each function with a LangChain Tool
search_tool = Tool(
    name="search_documents",
    description="Searches a knowledge base for relevant documents.",
    func=search_documents,
    args_schema=SearchArgs
)

api_tool = Tool(
    name="call_external_api",
    description="Calls an external HTTP endpoint with a JSON payload.",
    func=call_external_api,
    args_schema=APICallArgs
)

format_tool = Tool(
    name="format_response",
    description="Formats a list of document dictionaries into a readable string.",
    func=format_response,
    args_schema=FormatArgs
)

# List of all available tools
available_tools = [search_tool, api_tool, format_tool]

from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate

SYSTEM_INSTRUCTION = """
You are an autonomous research assistant. Your goal is to answer the user's query by using the available tools.
- Use `search_documents` to retrieve relevant information.
- If you need external data, use `call_external_api`.
- After gathering data, call `format_response` to produce a human‑readable answer.
- Only output a final answer when you are confident that the response is complete.
- Do NOT fabricate information; always rely on tool output.
"""

system_msg = SystemMessagePromptTemplate.from_template(SYSTEM_INSTRUCTION)
human_msg = HumanMessagePromptTemplate.from_template("{input}")

prompt = ChatPromptTemplate(messages=[system_msg, human_msg])

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, FunctionMessage
from typing import List, Dict, Any

# Initialize the LLM with function calling enabled
llm = ChatOpenAI(
    model_name="gpt-4o-mini",          # or "gpt-4o" for higher quality
    temperature=0.0,                   # deterministic for debugging
    max_tokens=1024,
    openai_api_key=os.getenv("OPENAI_API_KEY"),
    # Pass the function definitions automatically from the tools
    functions=[tool.get_openai_function() for tool in available_tools],
    function_call="auto",              # let the model decide when to call
)

def run_autonomous_agent(user_query: str,
                         max_steps: int = 10) -> str:
    """
    Executes the autonomous loop.
    Returns the final answer string.
    """
    # Conversation buffer
    messages: List[Dict[str, Any]] = [
        {"role": "system", "content": SYSTEM_INSTRUCTION},
        {"role": "user", "content": user_query}
    ]

    for step in range(max_steps):
        # 1️⃣ Generate a response
        response = llm(messages=messages)

        # 2️⃣ Check for function call
        if response.additional_kwargs.get("function_call"):
            func_name = response.additional_kwargs["function_call"]["name"]
            arguments_str = response.additional_kwargs["function_call"]["arguments"]
            arguments = json.loads(arguments_str)

            # Find the matching tool
            tool = next((t for t in available_tools if t.name == func_name), None)
            if not tool:
                raise ValueError(f"Tool {func_name} not registered.")

            # 3️⃣ Execute the tool
            tool_result = tool.run(**arguments)

            # 4️⃣ Append tool result as a function message
            messages.append({
                "role": "assistant",
                "content": None,
                "function_call": {
                    "name": func_name,
                    "arguments": arguments_str
                }
            })
            messages.append({
                "role": "function",
                "name": func_name,
                "content": json.dumps(tool_result)
            })
            # Continue to next iteration
        else:
            # Model gave a plain answer – treat as final
            final_answer = response.content
            return final_answer

    # If we exit the loop without a plain answer, fallback
    return "I'm unable to produce a definitive answer within the allotted steps."

if __name__ == "__main__":
    query = "What are the latest trends in renewable energy investment for 2024?"
    answer = run_autonomous_agent(query)
    print("\n=== Final Answer ===")
    print(answer)

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

def run_with_memory(user_query: str, max_steps: int = 10) -> str:
    # Load previous chat history into messages
    messages = [
        {"role": "system", "content": SYSTEM_INSTRUCTION},
        *memory.load_memory_variables({})["chat_history"]
    ]
    messages.append({"role": "user", "content": user_query})

    # The rest of the loop is identical to `run_autonomous_agent`
    # (omitted for brevity)
    # After final answer:
    memory.save_context({"input": user_query}, {"output": final_answer})
    return final_answer

def get_ticket_status(ticket_id: str) -> Dict:
    """
    Calls the internal ticketing system.
    """
    endpoint = f"https://support.example.com/api/tickets/{ticket_id}"
    return call_external_api(endpoint, {})

def summarize_resolution(steps: List[Dict]) -> str:
    """
    Turns raw resolution steps into a concise response.
    """
    formatted = "\n".join([f"{i+1}. {step['action']}" for i, step in enumerate(steps)])
    return f"Here is how you can resolve the issue:\n{formatted}"

User: My order #12345 hasn't shipped yet. What can I do?

import unittest

class TestTools(unittest.TestCase):
    def test_search_documents(self):
        results = search_documents("python testing")
        self.assertIsInstance(results, list)
        self.assertGreater(len(results), 0)

    def test_format_response_empty(self):
        self.assertEqual(format_response([]), "I couldn't find any relevant information.")

import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

def run_autonomous_agent(...):
    for step in range(max_steps):
        logging.info(f"Step {step+1}: Sending message to LLM")
        response = llm(messages=messages)
        logging.info(f"LLM response: {response}")

        if response.additional_kwargs.get("function_call"):
            logging.info(f"Function call detected: {response.additional_kwargs['function_call']}")
            # after execution
            logging.info(f"Function result: {tool_result}")
        else:
            logging.info("Final answer produced.")
            return response.content

from langchain.callbacks import LangChainTracer

tracer = LangChainTracer()
llm = ChatOpenAI(..., callbacks=[tracer])