Most LangChain tutorials stop at a toy example that prints a single string. To ship a real feature, you need to lcel compose prompt model parser into one pipeline that accepts structured input, calls an LLM, and returns typed output without hand-wiring intermediate steps.
Step 1: Define a parameterized prompt template
Start with a prompt that separates system instructions from user variables. Use ChatPromptTemplate from langchain_core.prompts. It compiles to a list of messages and validates missing variables at build time, not at runtime.
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "You are a senior Python reviewer. Reply with concise, actionable feedback."),
("user", "Review this function:\n\n{code}\n\nFocus on {focus}.")
])We keep the template strict: no f-string interpolation in Python, let LangChain handle escaping. This avoids injection of unintended braces and makes the template portable to a config file. If you need few-shot examples, append them as tuples before the final user message:
prompt = ChatPromptTemplate.from_messages([
("system", "You are a senior Python reviewer."),
("human", "def foo(): pass\nFocus on bugs."),
("ai", "Missing return; no input validation."),
("user", "Review this function:\n\n{code}\n\nFocus on {focus}."),
])Variables are declared inline; calling invoke with a missing key raises KeyError immediately. That fail-fast behavior beats discovering it after a paid model call.
Step 2: Configure the model binding
Use ChatOpenAI from langchain_openai. Pin the model, temperature, and timeouts. If you route through an OpenAI-compatible gateway such as n4n.ai, set base_url to its endpoint and use any of the 240+ addressed models without changing client code; it also applies automatic fallback when a provider is degraded.
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.1,
max_retries=2,
# base_url="https://api.n4n.ai/v1", # optional gateway
# api_key="your-key",
)Do not use legacy openai completions. Chat models give you message semantics and native tool calling. Set temperature low for review tasks; high values waste tokens on hallucinated nitpicks. For long inputs, set max_tokens explicitly to avoid silent truncation. If you need streaming, the same object supports .stream without extra config.
Step 3: Pick an output parser that matches your contract
The parser is not an afterthought. It defines the boundary between probabilistic text and your deterministic code. For free-form text, StrOutputParser strips the AIMessage wrapper. For structured data, use JsonOutputParser or PydanticOutputParser.
from langchain_core.output_parsers import StrOutputParser
parser = StrOutputParser()If you need JSON, bind a schema and parse:
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
class Review(BaseModel):
bugs: list[str] = Field(description="List of concrete bugs")
style: list[str] = Field(description="Style issues")
parser = JsonOutputParser(pydantic_object=Review)
prompt = prompt.partial(format_instructions=parser.get_format_instructions())The partial call bakes the schema instructions into the prompt so the model knows the exact shape to emit. Without it, you are begging for malformed JSON. For comma-separated values, CommaSeparatedListOutputParser exists, but prefer Pydantic when the contract is stable.
Step 4: lcel compose prompt model parser with the pipe operator
LCEL uses | to build a RunnableSequence. The expression prompt | model | parser is the canonical way to lcel compose prompt model parser into a single callable. Each segment is a Runnable, so you get .invoke, .stream, .batch, and .ainvoke for free.
chain = prompt | model | parser
result = chain.invoke({"code": "def add(a,b): return a+b", "focus": "edge cases"})
print(result)The pipe enforces left-to-right data flow: prompt renders the dict into messages, model returns an AIMessage, parser extracts the string or validated object. No LLMChain boilerplate, no ConversationChain state bugs. You can inspect the graph with chain.get_graph().print_ascii() to confirm the topology.
Why not legacy chains
Legacy Chain subclasses hide the runnable contract and break streaming. LCEL sequences are debuggable with RunnableLambda and RunnableParallel. You can swap any segment without touching the others. For example, replacing model with a local ChatOllama instance requires zero changes to prompt or parser.
Step 5: Run the chain and verify the output shape
Verification is part of the how-to. Write a small script that asserts the contract. For the string parser:
def test_string_chain():
out = chain.invoke({"code": "x=1", "focus": "typing"})
assert isinstance(out, str)
assert len(out) > 10
print("OK:", out[:80])For the Pydantic parser, the chain returns a Review instance:
pydantic_chain = prompt | model | JsonOutputParser(pydantic_object=Review)
review = pydantic_chain.invoke({
"code": "def f(): pass",
"focus": "bugs",
"format_instructions": parser.get_format_instructions()
})
assert isinstance(review, Review)
assert hasattr(review, "bugs")Success means the script exits zero and the printed output matches the expected type. If you see OutputParserException, the model ignored format instructions; lower temperature or add few-shot examples. Wire these tests into CI so a model swap cannot silently break your schema.
Step 6: Add retries and parallel enrichment
Production chains need resilience. Wrap the model with .with_retry() to handle transient 429s. Use RunnableParallel to compute side data while the LLM runs.
from langchain_core.runnables import RunnableParallel, RunnableLambda
enrich = RunnableLambda(lambda x: {"len": len(x["code"])})
combo = RunnableParallel({"review": chain, "meta": enrich})
out = combo.invoke({"code": "def add(a,b): return a+b", "focus": "edge cases"})
# out["review"] is parsed, out["meta"]["len"] is 27This pattern keeps the lcel compose prompt model parser core intact while adding orthogonal logic. The parallel branch never blocks the model call; both execute concurrently because RunnableParallel awaits independent runnables.
Step 7: Streaming and batch calls
LCEL shines when you need tokens incrementally. Call .stream on the string chain:
for chunk in chain.stream({"code": "x=1", "focus": "security"}):
print(chunk, end="", flush=True)Batch processes multiple inputs in one call:
batch_out = chain.batch([
{"code": "a=1", "focus": "style"},
{"code": "b=2", "focus": "bugs"},
])Both methods reuse the same composed runnable. You did not write a loop or async glue. Streaming returns chunks of the parsed output (for StrOutputParser it is strings; for JSON it may be partial text, so avoid streaming structured parsers unless you use an incremental JSON parser).
Step 8: Deployment and secrets hygiene
Put model credentials in environment variables, never in source. LangChain reads OPENAI_API_KEY automatically; for a gateway, set base_url via env and pass api_key=os.getenv("N4N_API_KEY"). Wrap chain construction in a factory function so tests can inject a fake model:
def build_chain(model=None):
model = model or ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
return prompt | model | StrOutputParser()This makes the lcel compose prompt model parser expression the single source of truth for your LLM feature, and lets you run unit tests against a FakeListChatModel in milliseconds.
Verification checklist
- Prompt renders without
KeyErroron required variables. - Model returns
AIMessagewithcontent. - Parser yields the exact type your downstream code imports.
chain.invokecompletes under your timeout.chain.streamyields incremental strings for text parsers.- CI test fails if the model returns malformed structured output.
If all hold, you have a maintainable LLM feature. The ability to lcel compose prompt model parser as a single expression reduces surface area for bugs and makes the data flow obvious to the next engineer who opens the file.