Bedrock with LangChain

Duration: 50 min

LangChain is a framework for building applications with language models. This module covers ChatBedrock, embeddings, chains, memory, and building production-grade AI applications.

What is LangChain?

LangChain provides abstractions for:

• Models — Unified interface to different LLMs
• Chains — Sequences of operations
• Memory — Conversation history management
• Agents — Autonomous decision-making
• Tools — Integration with external services

Installation

pip install langchain langchain-aws langchain-community

ChatBedrock

from langchain_aws import ChatBedrock
from langchain.schema import HumanMessage, AIMessage

# Initialize ChatBedrock
chat = ChatBedrock(
    model_id='anthropic.claude-3-sonnet-20240229-v1:0',
    region_name='us-east-1',
    model_kwargs={
        'temperature': 0.7,
        'max_tokens': 1024
    }
)

# Single message
response = chat.invoke([
    HumanMessage(content='What is AWS Bedrock?')
])
print(response.content)

# Multi-turn conversation
messages = [
    HumanMessage(content='What is machine learning?'),
    AIMessage(content='Machine learning is a subset of AI...'),
    HumanMessage(content='Give me an example')
]

response = chat.invoke(messages)
print(response.content)

Embeddings

from langchain_aws import BedrockEmbeddings

# Initialize embeddings
embeddings = BedrockEmbeddings(
    model_id='amazon.titan-embed-text-v2:0',
    region_name='us-east-1'
)

# Generate embedding for text
text = 'AWS Bedrock is a managed service'
embedding = embeddings.embed_query(text)
print(f'Embedding dimension: {len(embedding)}')

# Embed multiple texts
texts = [
    'AWS Bedrock',
    'Foundation models',
    'Machine learning'
]
embeddings_list = embeddings.embed_documents(texts)
print(f'Generated {len(embeddings_list)} embeddings')

Chains

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# Create a prompt template
prompt = PromptTemplate(
    input_variables=['topic'],
    template='Explain {topic} in simple terms for beginners.'
)

# Create a chain
chain = LLMChain(llm=chat, prompt=prompt)

# Run the chain
result = chain.run(topic='Artificial Intelligence')
print(result)

Conversation Memory

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# Create memory
memory = ConversationBufferMemory()

# Create conversation chain with memory
conversation = ConversationChain(
    llm=chat,
    memory=memory,
    verbose=True
)

# Multi-turn conversation
response1 = conversation.run(input='Hi, my name is Alice')
print(response1)

response2 = conversation.run(input='What is my name?')
print(response2)  # Model remembers "Alice"

# View conversation history
print(memory.buffer)

RAG with LangChain

from langchain.vectorstores import FAISS
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.chains import RetrievalQA

# Load documents
loader = TextLoader('company_docs.txt')
documents = loader.load()

# Split into chunks
splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

# Create vector store
vector_store = FAISS.from_documents(chunks, embeddings)

# Create RAG chain
qa_chain = RetrievalQA.from_chain_type(
    llm=chat,
    chain_type='stuff',
    retriever=vector_store.as_retriever(search_kwargs={'k': 5})
)

# Query
result = qa_chain.run('What is the vacation policy?')
print(result)

Agents with LangChain

from langchain.agents import Tool, initialize_agent, AgentType
from langchain.tools import DuckDuckGoSearchRun

# Define tools
search = DuckDuckGoSearchRun()

tools = [
    Tool(
        name='Search',
        func=search.run,
        description='Useful for searching the internet'
    ),
    Tool(
        name='Calculator',
        func=lambda x: str(eval(x)),
        description='Useful for math calculations'
    )
]

# Create agent
agent = initialize_agent(
    tools,
    chat,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# Run agent
result = agent.run('What is 2 + 2? Then search for AWS Bedrock pricing.')
print(result)

Custom Tools

from langchain.tools import tool
import boto3

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city"""
    # In real scenario, call weather API
    return f'Weather in {city}: Sunny, 72°F'

@tool
def query_database(query: str) -> str:
    """Query company database"""
    # In real scenario, query actual database
    return f'Query result: {query}'

# Use in agent
tools = [get_weather, query_database]

agent = initialize_agent(
    tools,
    chat,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

result = agent.run('What is the weather in New York?')
print(result)

Streaming with LangChain

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# Create chat with streaming
chat_streaming = ChatBedrock(
    model_id='anthropic.claude-3-sonnet-20240229-v1:0',
    region_name='us-east-1',
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# Invoke with streaming
response = chat_streaming.invoke([
    HumanMessage(content='Write a 500-word essay on AI')
])

Production Patterns

# ✅ Good: Error handling
from langchain.schema import LLMException

try:
    response = chat.invoke([HumanMessage(content=user_input)])
except LLMException as e:
    print(f'LLM error: {e}')
    return 'Sorry, I encountered an error.'

# ✅ Good: Input validation
def validate_input(text: str) -> bool:
    if len(text) > 10000:
        return False
    if len(text) < 1:
        return False
    return True

# ✅ Good: Response caching
from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()

# ✅ Good: Logging
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

logger.info(f'User input: {user_input}')
logger.info(f'Model response: {response}')