LLMs, prompt engineering, RAG systems, LangChain, and AI application development
sasmp_version
1.3.0
bonded_agent
06-ml-ai-engineer
bond_type
PRIMARY_BOND
skill_version
2.0.0
last_updated
2025-01
complexity
advanced
estimated_mastery_hours
180
prerequisites
["python-programming","machine-learning"]
unlocks
["deep-learning","mlops"]
LLMs & Generative AI
Production-grade LLM applications with prompt engineering, RAG systems, and modern AI development patterns.
Quick Start
# Production RAG System with LangChain (2024-2025)from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
# Initialize components
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# Document processing
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", ". ", " ", ""]
)
documents = text_splitter.split_documents(raw_documents)
# Vector store
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory="./chroma_db"
)
retriever = vectorstore.as_retriever(
search_type="mmr", # Maximum Marginal Relevance
search_kwargs={"k": 5, "fetch_k": 10}
)
# RAG chain
template = """Answer the question based only on the following context:
Context: {context}
Question: {question}
Answer thoughtfully and cite specific parts of the context."""
prompt = ChatPromptTemplate.from_template(template)
rag_chain = (
{: retriever, : RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
response = rag_chain.invoke()
(response)
"context"
"question"
# Query
"What are the key features?"
print
Core Concepts
1. Prompt Engineering Patterns
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
# System prompt design
system_prompt = """You are an expert data analyst assistant.
CAPABILITIES:
- Analyze data patterns and trends
- Generate SQL queries
- Explain statistical concepts
CONSTRAINTS:
- Only use information provided in the context
- Acknowledge uncertainty when relevant
- Format outputs in clear, structured way
OUTPUT FORMAT:
- Start with a brief summary
- Use bullet points for key findings
- Include confidence level (high/medium/low)
"""# Few-shot prompting
examples = [
{"input": "What's the average order value?",
"output": "```sql\nSELECT AVG(total_amount) as avg_order_value\nFROM orders\nWHERE status = 'completed';\n```"},
{"input": "Show top customers by revenue",
"output": "```sql\nSELECT customer_id, SUM(total_amount) as revenue\nFROM orders\nGROUP BY customer_id\nORDER BY revenue DESC\nLIMIT 10;\n```"}
]
example_prompt = ChatPromptTemplate.from_messages([
("human", "{input}"),
("ai", "{output}")
])
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_prompt=example_prompt,
examples=examples
)
# Chain of Thought prompting
cot_prompt = """Let's solve this step by step:
Question: {question}
Step 1: Identify the key components
Step 2: Break down the problem
Step 3: Apply relevant knowledge
Step 4: Synthesize the answer
Reasoning:"""# Self-consistency (multiple reasoning paths)asyncdefself_consistent_answer(question: str, n_samples: int = 5) -> str:
responses = await asyncio.gather(*[
llm.ainvoke(question) for _ inrange(n_samples)
])
# Majority voting or aggregationreturn aggregate_responses(responses)
import pytest
from unittest.mock import Mock, patch
from your_rag_system import RAGPipeline, DocumentProcessor
classTestRAGPipeline:
@pytest.fixturedefmock_llm(self):
llm = Mock()
llm.invoke.return_value = "Mocked response"return llm
@pytest.fixturedefrag_pipeline(self, mock_llm):
return RAGPipeline(llm=mock_llm)
deftest_retrieves_relevant_documents(self, rag_pipeline):
query = "What is machine learning?"
docs = rag_pipeline.retrieve(query)
assertlen(docs) > 0assertall("machine learning"in doc.page_content.lower()
for doc in docs[:3])
deftest_generates_grounded_response(self, rag_pipeline, mock_llm):
response = rag_pipeline.query("Test question")
mock_llm.invoke.assert_called_once()
assert response isnotNonedeftest_handles_empty_retrieval(self, rag_pipeline):
with patch.object(rag_pipeline.retriever, 'get_relevant_documents',
return_value=[]):
response = rag_pipeline.query("Obscure question")
assert"no information"in response.lower()
classTestDocumentProcessor:
deftest_chunks_documents_correctly(self):
processor = DocumentProcessor(chunk_size=100, chunk_overlap=20)
text = "A" * 250# 250 character document
chunks = processor.split(text)
assertlen(chunks) >= 2assertall(len(c) <= 100for c in chunks)
deftest_preserves_metadata(self):
processor = DocumentProcessor()
doc = Document(page_content="Test", metadata={"source": "test.pdf"})
chunks = processor.split_documents([doc])
assertall(c.metadata["source"] == "test.pdf"for c in chunks)
Best Practices
Prompt Engineering
# ✅ DO: Be specific and structured
prompt = """Task: Summarize the document.
Format: 3 bullet points
Constraints: Max 50 words per point
Tone: Professional"""# ✅ DO: Include examples# ✅ DO: Set clear output format# ✅ DO: Handle edge cases in prompt# ❌ DON'T: Vague prompts# ❌ DON'T: Assume LLM knows context# ❌ DON'T: Trust LLM output without validation