透過 7 個小型實作範例學習 LangChain：從聊天模型到 RAG

I've spent about 12 years building data pipelines on AWS, most recently for financial-crimes compliance in banking. When I started working with LLMs, I kept hitting the same wall. Every tutorial was either a one-line "hello world" or a 500-line agent framework I couldn't take apart.

So I did what data engineers do with any new tool: I broke it into the smallest pieces I could run, inspect, and reason about. The result is a repo of seven standalone LangChain scripts, each teaching exactly one concept. Every file runs on its own, and most are under 40 lines.

This post walks through them in order, from "call a model" to "answer questions over a PDF," and ends with how I'd move the whole thing onto Amazon Bedrock.

What LangChain is, and why it exists

LangChain is an open-source framework that puts standard, swappable interfaces around everything an LLM application needs besides the model itself.

A raw LLM is a function: text in, text out. A real application needs prompts built from variables, conversation state, output your code can parse, access to private data, and the ability to call functions. Without a framework, you write that glue yourself, once per provider.

As a data engineer, the last row is what sold me. LCEL treats every component as a pipeline stage with the same interface, which is exactly how I already think about ETL.

1. Chat models: one interface, any provider

The first script calls Llama 3.3 on Hugging Face. The OpenAI and Gemini versions sit right above it, commented out.
{% raw %}

#model = ChatOpenAI(model="gpt-5-mini")
#model = ChatGoogleGenerativeAI(model="gemini-2.5-flash")

llm = HuggingFaceEndpoint(
    repo_id="meta-llama/Llama-3.3-70B-Instruct",
    task="text-generation",
)
model = ChatHuggingFace(llm=llm)

response = model.invoke("tell me about ai agents in 10 words?")
print(response.content)

Swap the comment and nothing else changes. That one property is the whole value proposition of the framework in miniature.

2. Memory is just a list you resend

LLMs are stateless. Each call only knows what you send it, so "memory" means resending the conversation every turn.

conversation = [SystemMessage(content="You are a senior data engineer. Answer in 3 to 4 sentences.")]

while True:
    query = input("Ask a question: ")
    if query.lower() == "exit":
        break
    conversation.append(HumanMessage(content=query))
    response = model.invoke(conversation)
    conversation.append(AIMessage(content=response.content))
    print(response.content)

The lesson: history has a cost. The list grows every turn, so tokens and latency grow with it. Production apps trim or summarize older messages.

3. Chains with LCEL: pipelines for prompts

This is where LangChain clicked for me. The pipe operator connects components so each output feeds the next input:

prompt  = PromptTemplate(template="tell me about {topic} in 100 words")
prompt1 = PromptTemplate(template="based on below article, create 5 quiz questions\n\n{article}")
parser  = StrOutputParser()

chain = prompt | model | parser | prompt1 | model | parser
print(chain.invoke({"topic": "ai agents"}))

One input produces a 100-word article, which becomes the input for five quiz questions. Two details worth noticing:

• StrOutputParser turns the model's message object into a plain string.
• prompt1 has exactly one variable, so LangChain maps that string into {article} automatically. No glue code.

Every stage is a Runnable, so the same chain also supports .batch() and .stream() without changes.

4. Structured output: stop parsing with regex

If code consumes the answer, you want data, not prose. Define a Pydantic model and let LangChain enforce it:

class SQLOutput(BaseModel):
    query: str = Field(description="The generated SQL query.")
    explanation: str = Field(description="A brief explanation of the SQL query.")

model_structured = model.with_structured_output(SQLOutput)
response = model_structured.invoke("give me sql to get top 5 products by sales from orders table")

print(response.query)
print(response.explanation)

You get a typed object back. The Field descriptions are sent to the model too, so writing them well improves the output.

5. Tool calling, with guardrails a bank would accept

A tool is a Python function the model can ask to call. The @tool decorator reads the name, type hints, and docstring to build a schema the model understands.

I built tools over a local SQL Server retail database, plus a stock-price lookup:

Coming from regulated financial services, I wasn't going to let a model near a database without a gate. Every SQL tool goes through one validator:

def _validate_select(query: str) -> str:
    normalized = (query or "").strip()
    if not normalized:
        raise ValueError("The query is empty.")
    if normalized.endswith(";"):
        normalized = normalized[:-1].rstrip()
    if any(tok in normalized for tok in (";", "--", "/*", "*/")):
        raise ValueError("Only one SELECT statement is allowed.")
    if not re.match(r"^(SELECT|WITH)\b", normalized, flags=re.IGNORECASE):
        raise ValueError("Only SELECT queries are allowed.")
    if re.search(r"\b(INSERT|UPDATE|DELETE|MERGE|DROP|ALTER|CREATE|TRUNCATE|EXEC|EXECUTE|GRANT|REVOKE|INTO)\b",
                 normalized, flags=re.IGNORECASE):
        raise ValueError("Write/DDL keywords are not allowed.")
    return normalized

It blocks stacked statements, comments (a classic injection trick), anything that isn't SELECT or a CTE, and every write or DDL keyword, including SELECT ... INTO. It's a first line of defense, not the only one. In production, pair it with a read-only database login.

Then I bind the tools to the model:

model_tools = model.bind_tools([get_engine, get_stock_price, get_tables, get_schema])
result = model_tools.invoke("show me all the tables in the database")

for tool_call in result.tool_calls:
    print(f"Calling tool '{tool_call['name']}' with arguments {tool_call['args']}")

The most important thing I learned here: the model never runs your code. It returns a request, a tool name plus arguments. Your program decides whether to execute it. That boundary is where security and auditability live, and it's why tool calling is far less scary than it sounds.

6 and 7. RAG over an HR policy PDF

An LLM doesn't know your internal documents, and retraining it is expensive. Retrieval-Augmented Generation (RAG) retrieves the relevant passages at question time and hands them to the model as context. I used a sample bank HR policy as the source.

The pipeline has two halves:

1. Ingest (run once): PDF → Markdown → chunks → embeddings → Chroma
2. Retrieve (per question): question → embedding → nearest chunks

Ingest

embeddings = OpenAIEmbeddings(model="text-embedding-3-small", dimensions=300)

loader = PyMuPDF4LLMLoader("docs/Bank_HR_Policy.pdf", mode="single")
docs = loader.load()

splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[
    ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"),
])
chunks = splitter.split_text(docs[0].page_content)

vector_store = Chroma(collection_name="hr_policies",
                      embedding_function=embeddings,
                      persist_directory="./chroma_db")
vector_store.add_documents(chunks)

The choice that mattered most was how to split. Splitting by a fixed character count is the common default, and it can cut a policy rule in half. My earlier experiment is still in the repo as chunking.py for comparison. Converting the PDF to Markdown first and splitting on headings means each chunk is one complete policy section. The heading text is stored as metadata, so every result can say where it came from.

I also set the embedding size to 300 dimensions. That's smaller and cheaper to store and search, and plenty for a single policy document.

Retrieve

embeddings = OpenAIEmbeddings(model="text-embedding-3-small", dimensions=300)  # must match ingest
vector_store = Chroma(collection_name="hr_policies",
                      embedding_function=embeddings,
                      persist_directory="./chroma_db")

similar_docs = vector_store.similarity_search("What is the policy on remote work?", k=2)

The question must be embedded with the same model and dimensions used at ingest, or the vectors aren't comparable. The search returns the two chunks closest in meaning, even when they share no exact words with the question.

One gotcha: add_documents() appends. Run ingestion twice and you get duplicate chunks. Delete ./chroma_db before a rebuild.

Taking it to AWS with Amazon Bedrock

The repo uses OpenAI, but because LangChain abstracts the provider, moving to Amazon Bedrock is mostly a two-line change with the langchain-aws package:

from langchain_aws import ChatBedrockConverse, BedrockEmbeddings

model = ChatBedrockConverse(model="<a Bedrock model ID enabled in your account>")
embeddings = BedrockEmbeddings(model_id="amazon.titan-embed-text-v2:0")

The chains, structured output, tool binding, and Chroma code stay the same. For my world, the bigger wins are operational:

• No API keys in .env. Bedrock authenticates with IAM roles, so access is governed and auditable like any other AWS service.
• Data stays in your AWS account and Region, which matters a lot in banking.
• A managed RAG path. Bedrock Knowledge Bases can replace the hand-built ingest and retrieve steps when you outgrow a local Chroma store.

One thing to re-check when you switch: embedding dimensions. Titan Text Embeddings V2 supports a fixed set of sizes, not arbitrary values like the 300 I used with OpenAI, and you must re-ingest whenever the embedding model changes.

What I learned

• Small scripts beat big frameworks for learning. One concept per file made each abstraction obvious before I combined them.
• LCEL is just a pipeline. If you've built ETL, you already understand chains.
• The model proposes; your code disposes. Tool calling is safe to the degree your validation and permissions are.
• RAG quality starts at chunking. Structure-aware splitting mattered more than anything I did at query time.
• Load config before imports that read it. My tools.py reads environment variables at import time, so calling load_dotenv() after import tools silently used defaults. Small bug, real lesson.

What's next

Each script stops at a clear boundary, and each next step is a post of its own:

• Close the RAG loop: feed retrieved chunks into a prompt so the model answers with citations.
• Add an agent loop: execute tool calls, return results, and repeat until the model answers.
• Text-to-SQL agent: chain get_tables → get_schema → validate_select_query → run_query so plain-English questions return real data.
• Run it all on Bedrock, with Knowledge Bases for retrieval.

Try it yourself

The full repo, with setup steps and a concept-by-concept README, is here: https://github.com/KiranGunturu/langchain/tree/main

Clone it, add an OPENAI_API_KEY to a .env file, and run any script on its own. If you're a data engineer getting into GenAI, I'd love to hear which concept clicked for you, or what you'd want covered next.