📝 Text Summarizer#

Text Summarizer introduces LangChain and its declarative composition syntax, LCEL (LangChain Expression Language). Rather than manually formatting prompt strings, calling the raw SDK, and unwrapping nested response objects in imperative code, this service links three modular runnables into a pipeline using the Unix-style pipe operator (chain = prompt | model | parser).

When you submit a long passage of text, the chain injects it into a ChatPromptTemplate placeholder ({data}), sends the rendered prompt to ChatOpenAI at a low temperature (0.3) so the summary stays faithful to the source text, and passes the resulting AIMessage through StrOutputParser() to return a clean Python string.

1. Composable Pipelines with LCEL (|)

In LangChain, every component — prompts, models, output parsers, retrievers — implements the unified Runnable interface (.invoke(), .batch(), .stream(), .ainvoke()). Overloading Python's bitwise OR operator (|) chains runnables into a RunnableSequence where the output of step N automatically becomes the input to step N + 1:

1. ChatPromptTemplate: accepts {"data": long_text} and outputs a formatted PromptValue.
2. ChatOpenAI: accepts the PromptValue, calls the LLM API, and outputs an AIMessage(content=..., response_metadata=...).
3. StrOutputParser: extracts ai_message.content as a plain str ready for the caller.

2. Separation of Prompt Template and Runtime Data

Using ChatPromptTemplate.from_template("...{data}") instead of Python f-strings keeps prompt templates reusable, inspectable, and safe from accidental brace-escaping bugs when user text itself contains curly braces (such as JSON snippets or code).

3. Controlling Factual Faithfulness with Temperature

Summarization is a convergent task: we want the model to compress facts already present in the input, not invent creative embellishments. Setting temperature=0.3 sharpens the token probability distribution so the model strongly favors high-probability, faithful phrasings over speculative tokens.

4. Import-Safe Client Initialization

Passing api_key=os.getenv("OPENAI_API_KEY") or "missing-key" allows the module to be imported by the Flask server at startup even in environments where a specific provider key is not yet configured, deferring authentication errors until the endpoint is actually called.

Core AI Module (src/python/01_langchain_summarizer.py)

src/python/01_langchain_summarizer.py implements a declarative text summarization pipeline using LangChain Expression Language (LCEL). It composes a parameterized prompt template, a low-temperature chat model, and a string output parser into a single runnable sequence (prompt | model | parser).

Schemas, LCEL Pipeline Components, and Service Function

• TEXT_LANGCHAIN: str: A multi-sentence sample passage describing LangChain's role in building AI workflows, used as the default input text.
• class SummarizerRequest(BaseModel) & class SummarizerResponse(BaseModel): Pydantic schemas defining the input payload (text: str = TEXT_LANGCHAIN) and output payload (text: str, summary: str).
• LCEL Chain Construction and Execution:
  1. Step ① — Create a Reusable Prompt Template: Defines prompt = ChatPromptTemplate.from_template("Give a short, friendly summary of this long text:\n\n{data}"), exposing {data} as a named input variable.
  2. Step ② — Initialize the LangChain Chat Model: Instantiates model = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), api_key=os.getenv("OPENAI_API_KEY") or "missing-key", base_url=os.getenv("OPENAI_BASE_URL") or None, temperature=0.3, timeout=20.0). Setting temperature=0.3 keeps the generated summary factually grounded in the source passage.
  3. Step ③ — Initialize the Output Parser: Creates parser = StrOutputParser() to unwrap the model's AIMessage response object into a plain Python str.
  4. Step ④ — Compose the Pipeline with LCEL (|): Links chain = prompt | model | parser into a unified RunnableSequence where each stage's output automatically feeds into the next stage's input.
  5. Step ⑤ — Invoke the Chain in summarize(long_text: str = TEXT_LANGCHAIN) -> str: Calls return chain.invoke({"data": long_text}), binding long_text to the {data} placeholder and returning the parsed summary string.

How the AI Components Connect End-to-End

1. summarize(long_text) receives the passage to condense and passes {"data": long_text} into chain.invoke().
2. ChatPromptTemplate substitutes long_text into the {data} slot to produce a formatted PromptValue.
3. ChatOpenAI sends the rendered prompt to gpt-4o-mini at temperature=0.3 and returns an AIMessage containing the summary.
4. StrOutputParser extracts AIMessage.content as a plain string and returns it to fulfill the SummarizerResponse contract.

LangChain Text Summarizer Service.

"""LangChain Text Summarizer Service.

This service demonstrates prompt chaining using LangChain LCEL (LangChain Expression Language):
1. Loads API credentials and model configuration from environment variables (.env).
2. Sets up a ChatPromptTemplate with a data placeholder.
3. Chains ChatPromptTemplate | ChatOpenAI | StrOutputParser using the LCEL pipe operator (|).
4. Uses Pydantic models (SummarizerRequest, SummarizerResponse) for structured serialization.
5. Invokes the chain with input text to produce a concise, friendly summary.
"""

import os

from dotenv import find_dotenv, load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

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

TEXT_LANGCHAIN = (
    "Information about LangChain and its usage in building AI applications. "
    "We will explore how it simplifies the integration of language models into "
    "various applications. It provides a structured way to create prompts, manage "
    "outputs, and chain together multiple language model calls. This allows "
    "developers to build complex AI-driven workflows efficiently."
)


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class SummarizerRequest(BaseModel):
    """Input payload for text summarization queries."""

    text: str = TEXT_LANGCHAIN


class SummarizerResponse(BaseModel):
    """Output payload containing the original text and generated summary."""

    text: str
    summary: str


# ---------------------------------------------------------------------------
# LangChain LCEL Chain Definition
# ---------------------------------------------------------------------------
# ① a reusable prompt with a {data} blank
prompt = ChatPromptTemplate.from_template(
    "Give a short, friendly summary of this long text:\n\n{data}"
)

# ② the same model from Class 1, wrapped for LangChain
#   (placeholder key avoids an import-time crash when OPENAI_API_KEY is unset)
model = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY") or "missing-key",
    base_url=os.getenv("OPENAI_BASE_URL") or None,
    temperature=0.3,
    timeout=20.0,
)

# ③ the package-opener: hands back plain text
#   (StrOutputParser unwraps the AIMessage into a plain string)
parser = StrOutputParser()

# ④ snap them together  (| means "then")
chain = prompt | model | parser


# ---------------------------------------------------------------------------
# Core Service Logic
# ---------------------------------------------------------------------------
def summarize(long_text: str = TEXT_LANGCHAIN) -> str:
    """Summarize long text into a concise, friendly summary using LangChain.

    Args:
        long_text: Text content to summarize.

    Returns:
        Concise summary string generated by the LLM.
    """
    # ⑤ run it; the dict fills the {data} blank by name
    return chain.invoke({"data": long_text})

Flask Blueprint server and route handler (summarizer_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 core LangChain and LLM engineering concepts demonstrated by the Text Summarizer service:

1. LangChain Expression Language (LCEL): The pipe operator (chain = prompt | model | parser) declaratively composes modular objects into a single RunnableSequence where each step's output feeds into the next step's input.
2. The Unified Runnable Interface: Every LCEL component (prompts, chat models, output parsers) exposes standard execution methods (.invoke(), .batch(), .stream(), .ainvoke()) for synchronous, parallel, or streaming execution.
3. Parameterized Prompt Templates: ChatPromptTemplate.from_template("...{data}") separates static prompt instructions from runtime user text and safely substitutes variables by dictionary key without raw Python f-string brace-escaping bugs.
4. Low Temperature for Convergent Tasks: Setting temperature=0.3 on ChatOpenAI sharpens token probabilities so the model favors faithful compression of source facts rather than creative speculation.
5. Unwrapping Responses with StrOutputParser: Instead of manually indexing response.choices[0].message.content, StrOutputParser() automatically converts the model's AIMessage envelope into a clean Python str.
6. Import-Safe API Key Fallback: Supplying api_key=os.getenv("OPENAI_API_KEY") or "missing-key" lets Flask import the service module at startup even before a key is configured, deferring authentication checks until chain.invoke() is called.
7. Pydantic Request/Response Contracts: SummarizerRequest and SummarizerResponse document the expected JSON structure and provide a ready-to-run default passage (TEXT_LANGCHAIN).