Prompt Templates, Structured Output & Output Parsers — Part 7

prompt = f"You are a researcher. Task: {task}\nProduce numbered findings."
response = llm.invoke([HumanMessage(content=prompt)])
print(response.content)  # a raw string — could be anything

source langchain-env/bin/activate
pip install langchain-google-genai langchain-core pydantic python-dotenv

# create a file: 20_prompt_templates.py
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage
from langchain_google_genai import ChatGoogleGenerativeAI

load_dotenv()
llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)

# -------------------------------------------------------
# Basic template: named variables replace f-string interpolation
# -------------------------------------------------------
headline_template = ChatPromptTemplate.from_messages([
    ("system", """You are a senior editor at Tech News Daily.
Your job is to classify incoming news articles and extract clean metadata.
Be precise. Do not add commentary outside the requested fields."""),
    ("human", """Article title: {title}
Article body: {body}

Classify this article and extract: category, estimated reading time (minutes), and a 1-sentence summary.""")
])

# The template is now an object you can inspect, test, and store
print("Template variables:", headline_template.input_variables)
# → ['title', 'body']

# Invoke by passing variables — the template handles interpolation
chain = headline_template | llm

result = chain.invoke({
    "title": "Gemini 3.5 Flash Cuts Inference Costs by 40%",
    "body": "Google DeepMind announced today that Gemini 3.5 Flash achieves a 40% reduction in per-token inference costs compared to its predecessor, while maintaining benchmark parity on coding and reasoning tasks. The model is available immediately via the Gemini API."
})
print("\nRaw response (still a string):")
print(result.content)

python 20_prompt_templates.py

# Add this to 20_prompt_templates.py
# -------------------------------------------------------
# MessagesPlaceholder: injects a dynamic list of messages at a fixed slot
# This is how you add multi-turn history to any template
# -------------------------------------------------------
editor_template = ChatPromptTemplate.from_messages([
    ("system", "You are a senior editor at Tech News Daily. Be concise and direct."),
    MessagesPlaceholder(variable_name="chat_history"),  # injected at runtime
    ("human", "{current_question}")
])

# Simulate a multi-turn session
history = [
    HumanMessage(content="What is the headline policy for breaking news?"),
    # In a real session this would include the AI's response too
]

response = (editor_template | llm).invoke({
    "chat_history": history,
    "current_question": "Should breaking news headlines include question marks?"
})
print("\nEditor response:")
print(response.content)

# create a file: 21_structured_output.py
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI

load_dotenv()
llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)

# -------------------------------------------------------
# Step 1: Define the schema as a Pydantic model
# This becomes the "contract" between your agent and its callers
# -------------------------------------------------------
class Article(BaseModel):
    """Structured representation of a news article's metadata."""
    
    headline: str = Field(
        description="The article headline, max 12 words, no question marks"
    )
    category: Literal["ai", "cloud", "security", "hardware", "business"] = Field(
        description="The primary category of the article"
    )
    summary: str = Field(
        description="A single sentence (max 25 words) summarising the article"
    )
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="Your confidence in this classification, 0.0–1.0"
    )
    breaking: bool = Field(
        description="True if this is breaking news (announced in last 6 hours)"
    )

# -------------------------------------------------------
# Step 2: Bind the schema to the LLM
# Under the hood, LangChain chooses tool-calling vs JSON mode
# based on what the model supports — you do not need to decide
# -------------------------------------------------------
structured_llm = llm.with_structured_output(Article)

# -------------------------------------------------------
# Step 3: Build the chain — same pattern as before
# -------------------------------------------------------
classify_template = ChatPromptTemplate.from_messages([
    ("system", """You are a news classification engine for Tech News Daily.
Classify articles accurately. Confidence reflects how clearly the article fits one category.
If an article spans multiple categories, pick the most dominant one."""),
    ("human", """Article to classify:
Title: {title}
Body: {body}
Published: {published_at}""")
])

classify_chain = classify_template | structured_llm

# -------------------------------------------------------
# Test with three different articles
# -------------------------------------------------------
test_articles = [
    {
        "title": "AWS Announces 30% Price Cut on EC2 GPU Instances",
        "body": "Amazon Web Services reduced pricing on its P4d and P5 GPU instance families by 30% effective today, citing improved manufacturing economics and competition from Google Cloud and Azure in the AI training market.",
        "published_at": "2026-06-13T09:00:00Z"
    },
    {
        "title": "Critical Zero-Day Found in OpenSSH 9.x",
        "body": "Security researchers at Qualys disclosed a critical remote code execution vulnerability in OpenSSH versions 9.0–9.8. The CVE-2026-0412 flaw allows unauthenticated attackers to execute arbitrary code as root on vulnerable Linux systems. A patch is available.",
        "published_at": "2026-06-13T07:30:00Z"  # recent → breaking
    },
    {
        "title": "NVIDIA Unveils Blackwell Ultra B300 GPU Architecture",
        "body": "NVIDIA's new Blackwell Ultra architecture delivers 2.5x the FP8 throughput of H100 and introduces a new NVLink 5 interconnect. The B300 will ship in Q3 2026 with a focus on large-scale LLM inference clusters.",
        "published_at": "2026-06-12T14:00:00Z"
    }
]

print("=== Tech News Daily — Article Classifier ===\n")
for article in test_articles:
    result: Article = classify_chain.invoke(article)
    
    # result is a proper Python object — fully typed
    print(f"Title:      {article['title']}")
    print(f"Headline:   {result.headline}")
    print(f"Category:   {result.category}")
    print(f"Breaking:   {'🔴 YES' if result.breaking else 'No'}")
    print(f"Confidence: {result.confidence:.0%}")
    print(f"Summary:    {result.summary}")
    print("-" * 60)

python 21_structured_output.py

# create a file: 22_output_parsers.py
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.output_parsers import PydanticOutputParser
from langchain.output_parsers import OutputFixingParser
from langchain_google_genai import ChatGoogleGenerativeAI

load_dotenv()
llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)

# -------------------------------------------------------
# StrOutputParser: the simplest parser — strips metadata,
# returns just the string content. Use this for free-form output.
# -------------------------------------------------------
summary_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a copy editor. Write tight, punchy summaries."),
        ("human", "Summarise this in exactly 2 sentences: {text}")
    ])
    | llm
    | StrOutputParser()  # converts AIMessage to plain string
)

text = "The EU AI Act's enforcement provisions came into effect today, requiring all high-risk AI systems deployed in the EU to undergo mandatory conformity assessments. Companies have 6 months to comply or face fines of up to 3% of global annual revenue."
summary = summary_chain.invoke({"text": text})
print("Summary (str):", summary)
print("Type:", type(summary))  # → <class 'str'>

# -------------------------------------------------------
# PydanticOutputParser: instructs the model via prompt to
# produce JSON matching your schema, then parses it.
# Less reliable than with_structured_output() but works
# with any model, including those without tool calling.
# -------------------------------------------------------
class EditDecision(BaseModel):
    """An editorial decision on a news article."""
    publish: bool = Field(description="Whether to publish this article")
    reason: str = Field(description="One sentence explaining the decision")
    priority: Literal["top", "standard", "hold"] = Field(
        description="Editorial priority: top (homepage), standard, or hold"
    )

parser = PydanticOutputParser(pydantic_object=EditDecision)

editorial_template = ChatPromptTemplate.from_messages([
    ("system", "You are the editor-in-chief of Tech News Daily."),
    ("human", """Evaluate this article and make an editorial decision.

Article: {article}

{format_instructions}""")
]).partial(format_instructions=parser.get_format_instructions())

editorial_chain = editorial_template | llm | parser

decision: EditDecision = editorial_chain.invoke({
    "article": "AWS quietly removed two deprecated EC2 instance types from its pricing page. No announcement was made."
})

print(f"\nPublish: {decision.publish}")
print(f"Priority: {decision.priority}")
print(f"Reason: {decision.reason}")

# -------------------------------------------------------
# OutputFixingParser: wraps any parser and adds an automatic
# retry if the initial parse fails.
# -------------------------------------------------------
# Simulate a scenario where the model might return malformed output:
fixing_parser = OutputFixingParser.from_llm(
    parser=PydanticOutputParser(pydantic_object=EditDecision),
    llm=llm
)

# In production, wrap your parser with OutputFixingParser for any
# model that sometimes produces slightly malformed JSON.
# The fixing parser calls the LLM once more with the parse error
# and the bad output, asking it to correct the format.
print("\n✓ OutputFixingParser configured — will auto-retry on malformed output")
print("  Use it like: fixing_parser.parse(malformed_string)")

python 22_output_parsers.py

User submits raw text
       ↓
  classify_chain → Article (typed Pydantic object)
       ↓
  researcher_agent reads Article.category to focus its research
       ↓
  writer_agent uses Article.headline + Article.summary as the brief
       ↓
  notifier_agent formats based on Article.breaking flag
       ↓
  Output: a structured, auditable pipeline with typed data at every seam