LangChain Implementation Guide: Building AI Applications in Australia

LangChain provides a modular architecture that abstracts the complexity of building AI applications while maintaining flexibility for customisation. Understanding its core components is essential for effective implementation.

Core Components

LangChain Expression Language (LCEL)

LCEL is the declarative composition system that makes building chains intuitive:

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# Define the chain using LCEL pipe syntax
prompt = ChatPromptTemplate.from_template(
    "Summarise the following Australian business document: {document}"
)

chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()

# Execute the chain
result = chain.invoke({"document": document_text})

LCEL provides automatic streaming, batching, and async support without additional configuration.

Chains are the workhorses of LangChain applications, enabling you to combine multiple operations into cohesive workflows. Understanding chain patterns is crucial for building robust AI systems.

Sequential Chains

Sequential chains pass output from one step to the next:

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# Chain 1: Extract key points
extract_prompt = ChatPromptTemplate.from_template(
    "Extract key points from: {text}"
)

# Chain 2: Summarise key points
summarise_prompt = ChatPromptTemplate.from_template(
    "Summarise these key points for Australian executives: {key_points}"
)

model = ChatOpenAI(model="gpt-4")

# Combine into sequential chain
full_chain = (
    {"key_points": extract_prompt | model | StrOutputParser()}
    | summarise_prompt
    | model
    | StrOutputParser()
)

Parallel Chains with RunnableParallel

Execute multiple operations simultaneously for efficiency:

from langchain_core.runnables import RunnableParallel

# Define parallel analysis
parallel_chain = RunnableParallel(
    sentiment=sentiment_chain,
    entities=entity_extraction_chain,
    summary=summary_chain,
    compliance_check=compliance_chain
)

# All run concurrently
results = parallel_chain.invoke({"document": business_report})

Conditional Chains with RunnableBranch

Route execution based on conditions:

from langchain_core.runnables import RunnableBranch

# Route based on document type
router = RunnableBranch(
    (lambda x: x["type"] == "contract", contract_analysis_chain),
    (lambda x: x["type"] == "invoice", invoice_processing_chain),
    (lambda x: x["type"] == "email", email_response_chain),
    default_chain  # Fallback
)

Agents represent the most powerful capability in LangChain—systems that can reason, make decisions, and take actions autonomously. They're essential for building sophisticated AI assistants and automation systems.

Agent Architecture

Agents combine three key elements:

• LLM: The reasoning engine that decides what to do
• Tools: Actions the agent can take
• Memory: Context from previous interactions

Creating Custom Tools

from langchain_core.tools import tool
from typing import Optional

@tool
def search_australian_business_registry(
    company_name: str,
    abn: Optional[str] = None
) -> str:
    """
    Search the Australian Business Registry for company information.
    Use this when you need to verify Australian business details.

    Args:
        company_name: The name of the company to search
        abn: Optional ABN for more precise lookup
    """
    # Implementation
    result = abr_api.search(company_name, abn)
    return f"Company: {result.name}, ABN: {result.abn}, Status: {result.status}"

@tool
def calculate_gst(amount: float, inclusive: bool = True) -> str:
    """
    Calculate GST for Australian transactions.

    Args:
        amount: The dollar amount
        inclusive: Whether the amount includes GST
    """
    if inclusive:
        gst = amount / 11
        net = amount - gst
    else:
        gst = amount * 0.1
        net = amount
    return f"GST: ${gst:.2f}, Net: ${net:.2f}, Total: ${net + gst:.2f}"

Building the Agent

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Define tools
tools = [
    search_australian_business_registry,
    calculate_gst,
    send_email,
    query_database
]

# Create agent prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an Australian business assistant.
    Use Australian English spelling.
    Always verify ABNs before processing transactions.
    Ensure GST compliance for all calculations."""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# Create the agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = create_tool_calling_agent(llm, tools, prompt)

# Create executor with error handling
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=10
)

Agent Execution Strategies

RAG systems combine the power of language models with external knowledge, enabling accurate, up-to-date responses grounded in your organisation's data. This is essential for building AI that understands your specific business context.

RAG Architecture Overview

Complete RAG Implementation

from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. Load documents
loader = DirectoryLoader(
    "./company_docs",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader
)
documents = loader.load()

# 2. Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["

", "
", ". ", " ", ""]
)
chunks = text_splitter.split_documents(documents)

# 3. Create embeddings and vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. Create retriever
retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximum marginal relevance
    search_kwargs={"k": 5, "fetch_k": 10}
)

# 5. Build RAG chain
template = """You are an Australian business assistant. Answer based on the context.
If the answer isn't in the context, say you don't have that information.

Context: {context}

Question: {question}

Answer:"""

prompt = ChatPromptTemplate.from_template(template)

def format_docs(docs):
    return "

".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4")
    | StrOutputParser()
)

# Query the system
response = rag_chain.invoke("What is our leave policy?")

Advanced RAG Techniques

Memory enables AI applications to maintain context across interactions, creating natural conversational experiences. LangChain provides multiple memory types for different use cases.

Memory Types

Implementing Conversation Memory

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Store for session histories
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# Create prompt with history placeholder
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an Australian business assistant. Be helpful and concise."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# Create chain
chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()

# Wrap with message history
conversational_chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# Use with session tracking
response = conversational_chain.invoke(
    {"input": "What are the GST requirements for my business?"},
    config={"configurable": {"session_id": "user_123"}}
)

Production Memory with Redis

from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_redis_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379",
        key_prefix="chat:",
        ttl=3600  # 1 hour expiry
    )

# Now conversations persist across restarts
conversational_chain = RunnableWithMessageHistory(
    chain,
    get_redis_history,
    input_messages_key="input",
    history_messages_key="history"
)

Moving LangChain applications to production requires careful attention to performance, reliability, and observability. These patterns ensure your AI systems perform well under real-world conditions.

Async Operations for Scale

import asyncio
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")

# Async invocation
async def process_documents(documents: list[str]) -> list[str]:
    tasks = [llm.ainvoke(doc) for doc in documents]
    return await asyncio.gather(*tasks)

# Streaming for better UX
async def stream_response(query: str):
    async for chunk in chain.astream({"input": query}):
        yield chunk

Error Handling and Retries

from langchain_core.runnables import RunnableConfig
from tenacity import retry, stop_after_attempt, wait_exponential

# Built-in retry configuration
llm = ChatOpenAI(
    model="gpt-4",
    max_retries=3,
    request_timeout=30
)

# Custom retry logic for chains
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def invoke_with_retry(chain, input_data):
    return await chain.ainvoke(input_data)

# Fallback chains
from langchain_core.runnables import RunnableWithFallbacks

primary_chain = prompt | ChatOpenAI(model="gpt-4")
fallback_chain = prompt | ChatOpenAI(model="gpt-3.5-turbo")

robust_chain = primary_chain.with_fallbacks([fallback_chain])

Observability with LangSmith

import os

# Enable LangSmith tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-australian-app"

# All chain executions are now traced
# View at smith.langchain.com

Caching for Cost Optimisation

from langchain_community.cache import RedisCache
from langchain_core.globals import set_llm_cache
import redis

# Configure Redis cache
redis_client = redis.Redis.from_url("redis://localhost:6379")
set_llm_cache(RedisCache(redis_client))

# Identical prompts now return cached responses
# Reduces API costs significantly

A Melbourne-based financial services firm implemented LangChain to build an intelligent compliance assistant, demonstrating the power of combining multiple LangChain capabilities.

# Compliance-specific tools
@tool
def check_disclosure_requirements(product_type: str) -> str:
    """Check ASIC disclosure requirements for a product."""
    # Query regulations database
    return requirements

@tool
def verify_client_suitability(client_id: str, recommendation: str) -> str:
    """Verify recommendation suits client's risk profile."""
    # Cross-reference client data
    return suitability_assessment

# Agent with compliance focus
compliance_agent = create_tool_calling_agent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=[check_disclosure_requirements, verify_client_suitability, ...],
    prompt=compliance_prompt
)

Key Implementation Lessons

1. Start with RAG: Ground responses in authoritative sources before adding agent capabilities
2. Validate rigorously: Financial compliance requires extensive testing and human review
3. Maintain audit trails: Every AI decision must be traceable for regulatory purposes
4. Plan for updates: Regulations change—build processes for keeping knowledge current