🌦️ LangChain Agent#

LangChain Single-Tool Agent implements the exact same single-tool agent pattern as the previous demo, but replaces 40 lines of manual JSON Schema definitions and two-phase loop plumbing with LangChain's high-level create_agent() factory.

You simply write a plain Python function — def get_weather(city: str) -> str: with a clear docstring — and pass tools=[get_weather] to create_agent(). LangChain automatically introspects the function signature to build the tool schema, manages the multi-turn tool_calls conversation history internally, and returns the complete message trace so the service can report both the final answer and a boolean tool_called flag.

1. Automatic Schema Introspection from Python Type Hints

Instead of hand-authoring a nested {"type": "function", "parameters": ...} dictionary, LangChain inspects your Python function at registration time:

• Function Name (get_weather): Becomes the tool identifier.
• Type Annotations (city: str): Compiled into JSON Schema property types ({"city": {"type": "string"}}) and required parameter lists.
• Docstring ("Get the current weather for a given city."): Becomes the tool description the LLM reads to decide whether to call the tool.

2. Automated Agent Loop Orchestration

When agent.invoke({"messages": [...]}) runs, LangChain's runtime graph handles the entire cycle behind the scenes: sending the prompt + tool schemas, checking if the model emitted tool_calls, executing get_weather(), packaging the output into a ToolMessage with the matching tool_call_id, and re-querying the LLM until it produces a final AIMessage.

3. Inspecting the Execution Trace

agent.invoke() returns a dictionary containing the full list of LangChain message objects accumulated during the turn: HumanMessage, AIMessage (with tool_calls), ToolMessage (with m.type == "tool"), and the final AIMessage. Checking any(m.type == "tool" for m in messages) provides a deterministic signal of whether the agent used the tool or answered from memory.

4. Config-Driven Provider Switching

By reading f"{prefix}_MODEL", f"{prefix}_API_KEY", and f"{prefix}_BASE_URL" based on PROVIDER = "groq", the agent runs on Groq's fast free tier by default while remaining switchable to OpenAI or Gemini via a single constant.

Core AI Module (src/python/04_langchain_one_tool.py)

src/python/04_langchain_one_tool.py builds a single-tool ReAct agent using LangChain's create_agent() factory and ChatOpenAI configured for Groq. Instead of hand-writing JSON Schema definitions and manual tool loops, it registers a plain Python function (get_weather) and lets LangChain introspect its signature and orchestrate the tool-calling cycle automatically.

Schemas, Tool Function, and Agent Execution

• Provider Selection (PROVIDER = "groq", prefix = PROVIDER.upper()): Selects the active model provider so GROQ_MODEL, GROQ_API_KEY, and GROQ_BASE_URL configure the OpenAI-compatible chat client.
• class LangchainOneToolRequest(BaseModel) & class LangchainOneToolResponse(BaseModel): Pydantic schemas defining the input query (message: str = "What is the weather like in New York?") and output contract (message: str, response: str, and tool_called: bool).
• get_weather(city: str) -> str: The plain Python tool function. LangChain inspects its name (get_weather), parameter type annotation (city: str), and docstring ("""Get the current weather for a given city.""") to auto-generate the tool schema sent to the LLM, returning f"The weather in {city} is too hot to handle, 28 degrees C." when invoked.
• Agent Construction and run_langchain_one_tool(question: str = "What is the weather like in New York?") -> tuple[str, bool]:
  1. Step ① — Initialize the LLM Client: Instantiates llm = ChatOpenAI(model=os.getenv(f"{prefix}_MODEL"), api_key=os.getenv(f"{prefix}_API_KEY") or "missing-key", base_url=os.getenv(f"{prefix}_BASE_URL") or None, temperature=0, timeout=20.0). Setting temperature=0 ensures deterministic tool selection and argument extraction.
  2. Step ② — Create the LangChain Tool-Calling Agent: Calls agent = create_agent(model=llm, tools=[get_weather], system_prompt=...) to compile the ReAct agent graph with instructions to invoke available tools when needed and state the final result clearly.
  3. Step ③ — Invoke the Agent Graph: Runs messages = agent.invoke({"messages": [{"role": "user", "content": question}]})["messages"], which sends the question and tool schema to the LLM, executes get_weather(city) if requested, feeds the ToolMessage back to the model, and returns the complete message trace.
  4. Step ④ — Detect Tool Execution in the Trace: Evaluates tool_called = any(m.type == "tool" for m in messages) to determine whether a ToolMessage was recorded during the turn.
  5. Step ⑤ — Return Final Reply and Indicator: Returns messages[-1].content, tool_called, where messages[-1] is the assistant's final AIMessage.

How the AI Components Connect End-to-End

1. run_langchain_one_tool(question) passes the user's question into agent.invoke({"messages": [...]}).
2. The compiled create_agent graph sends the system prompt, user message, and introspected get_weather schema to the Groq LLM.
3. When the question asks about weather, the model emits a tool_calls request; LangChain automatically runs get_weather(city), wraps the returned string in a ToolMessage, and re-invokes the LLM to synthesize the final answer (or skips the tool step for general-knowledge questions).
4. The message trace is inspected for any m.type == "tool" entry, and (messages[-1].content, tool_called) is returned to fulfill LangchainOneToolResponse.

Langchain Version V1 - Agents (single tool)

"""
Langchain Version V1 - Agents (single tool)

Teaching idea: give the agent ONE tool (get_weather), and show how LangChain's
`create_agent` automatically inspects tool signatures, triggers the tool
when needed (e.g. weather queries), and answers general queries directly.
"""

import os

from dotenv import find_dotenv, load_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

# ---------------------------------------------------------------------------
# Configuration & Provider Selection
# ---------------------------------------------------------------------------
# Load API keys and model names from .env
load_dotenv(find_dotenv())

# Pick a provider: "groq" (free + fast), "gemini" (free tier) or "openai" (needs billing).
# Its settings are read from <PROVIDER>_MODEL / <PROVIDER>_API_KEY / <PROVIDER>_BASE_URL in .env
# (get keys at https://console.groq.com or https://aistudio.google.com)
PROVIDER = "groq"
prefix = PROVIDER.upper()


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class LangchainOneToolRequest(BaseModel):
    """Input payload for single-tool LangChain agent queries."""

    message: str = "What is the weather like in New York?"


class LangchainOneToolResponse(BaseModel):
    """Output payload containing the final response and tool call indicator."""

    message: str
    response: str
    tool_called: bool


# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------
# Type hints (city: str) tell the model WHAT arguments to pass;
# the docstring tells the model WHEN to invoke this tool.
def get_weather(city: str) -> str:
    """Get the current weather for a given city."""
    return f"The weather in {city} is too hot to handle, 28 degrees C."


# ---------------------------------------------------------------------------
# Build the agent
# ---------------------------------------------------------------------------
# ① Initialize the LLM client pointed at the configured provider
#   (placeholder key avoids an import-time crash when the key is unset)
llm = ChatOpenAI(
    model=os.getenv(f"{prefix}_MODEL"),
    api_key=os.getenv(f"{prefix}_API_KEY") or "missing-key",
    base_url=os.getenv(f"{prefix}_BASE_URL") or None,
    temperature=0,
    timeout=20.0,
)

# ② Create the LangChain tool-calling agent with get_weather and system instructions
agent = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt=(
        "You are a helpful assistant. Use the available tools when needed. "
        "Always state the final result clearly and completely in your answer."
    ),
)


# ---------------------------------------------------------------------------
# Core Service Logic
# ---------------------------------------------------------------------------
def run_langchain_one_tool(question: str = "What is the weather like in New York?") -> tuple[str, bool]:
    """Run a user question through the single-tool LangChain agent."""
    # ③ Invoke the LangChain agent with the user's question
    messages = agent.invoke({"messages": [{"role": "user", "content": question}]})["messages"]

    # ④ The agent called the tool if any message in the trace has type "tool"
    tool_called = any(m.type == "tool" for m in messages)

    # ⑤ The last message is the final assistant reply
    return messages[-1].content, tool_called

Flask Blueprint server and route handler (langchain_one_tool_service) in src/python/app.py.

"""Flask server for 02-langchain-agents."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_summarize = _load_service("01_langchain_summarizer.py", "langchain_summarizer")
mod_memory = _load_service("02_langchain_memory_chat.py", "langchain_memory_chat")
mod_agent = _load_service("03_agent_one_tool.py", "agent_one_tool")
mod_lc_tool = _load_service("04_langchain_one_tool.py", "langchain_one_tool")
mod_lc_multi = _load_service("05_langchain_multi_tools.py", "langchain_multi_tools")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/summarize", methods=["POST"])
def summarizer_service():
    # ① Parse input text from JSON payload and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    text = (data.get("text") or data.get("message") or "").strip()
    if not text:
        return jsonify({"error": "Text cannot be empty"}), 400
    try:
        # ② Run the LangChain LCEL summarizer chain in 01_langchain_summarizer.py
        summary = mod_summarize.summarize(text)
        return jsonify({"result": summary, "summary": summary})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/memory-chat", methods=["POST"])
def memory_chat_service():
    # ① Parse user message and prior conversation history from JSON payload
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    history = data.get("history") or []
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Convert history dicts to ChatMessage models and run chat_with_memory()
        chat_history = [
            mod_memory.ChatMessage(**m) if isinstance(m, dict) else m
            for m in history
        ]
        reply, updated_history = mod_memory.chat_with_memory(message, chat_history)
        return jsonify({
            "result": reply,
            "response": reply,
            "history": [m.model_dump() if hasattr(m, "model_dump") else m for m in updated_history],
        })
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/agent", methods=["POST"])
def agent_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the raw OpenAI tool-calling loop in 03_agent_one_tool.py
        reply = mod_agent.run_agent_with_tool(message)
        return jsonify({"result": reply, "response": reply})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/langchain-one-tool", methods=["POST"])
def langchain_one_tool_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the LangChain single-tool agent in 04_langchain_one_tool.py
        reply, tool_called = mod_lc_tool.run_langchain_one_tool(message)
        return jsonify({"result": reply, "response": reply, "tool_called": tool_called})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/multi-tool-agent", methods=["POST"])
def multi_tool_agent_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the LangGraph ReAct multi-tool agent in 05_langchain_multi_tools.py
        reply, tool_called = mod_lc_multi.run_multi_tool_agent(message)
        return jsonify({"result": reply, "response": reply, "tool_called": tool_called})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Use this checklist to review the LangChain agent concepts and design patterns demonstrated by the LangChain Single-Tool Agent service:

1. Zero-Boilerplate Tool Registration: Passing a plain Python function (tools=[get_weather]) to create_agent() eliminates manual JSON Schema authoring by introspecting the function's name, type annotations (city: str), and docstring.
2. Docstrings as Semantic Routing Rules: The LLM reads get_weather.__doc__ ("Get the current weather for a given city.") to decide whether a user's prompt warrants calling the tool or answering directly.
3. Automated ReAct Loop via create_agent(): Calling agent.invoke({"messages": [...]}) automatically orchestrates the multi-turn tool loop — parsing tool_calls, running the Python function, appending ToolMessage objects with matching tool_call_ids, and re-prompting the LLM.
4. Deterministic Tool Calling with temperature=0: Configuring ChatOpenAI(..., temperature=0) minimizes sampling randomness when the model selects tools and formats JSON arguments.
5. Trace Inspection for Observability: Scanning the returned message list with any(m.type == "tool" for m in messages) provides a reliable boolean signal (tool_called) that the UI can display as an execution badge.
6. Prefix-Driven Provider Switching: Computing prefix = PROVIDER.upper() lets the service switch between Groq, Gemini, and OpenAI credentials in .env by changing a single constant.
7. Explicit Result-Stating System Prompt: Instructing the agent to "Always state the final result clearly and completely in your answer" ensures the model repeats the tool's output in its final AIMessage for the user.