Large Language Models like GPT-4 and Claude have revolutionized what's possible with AI. Our LLM integration services help you leverage these powerful models to build intelligent features that understand context, generate content, answer questions, and automate knowledge work. We implement Retrieval Augmented Generation (RAG) to ground responses in your data, fine-tune models for your specific domain, engineer prompts for optimal performance, and optimize costs for production use. Whether you need a customer support assistant, content generation system, code helper, or document analysis tool, we build production-ready LLM applications that deliver real business value.
SaaS companies adding AI features, enterprises automating knowledge work, content platforms scaling creation, legal firms analyzing documents, customer support teams enhancing agents, development teams building code assistants, e-commerce businesses personalizing descriptions, and any organization with text-heavy workflows.
Complete integration with OpenAI GPT-4, Anthropic Claude, and other LLM providers. Async request handling, streaming responses, error handling, retry logic, and fallback strategies.
RAG pipelines that ground LLM responses in your data. Vector database integration (Pinecone, Weaviate, Chroma), semantic search, chunk optimization, and re-ranking for accurate retrieval.
Expert prompt design using few-shot learning, chain-of-thought, ReAct patterns, and system instructions. A/B testing framework to optimize prompts for accuracy, cost, and latency.
Fine-tune models on your proprietary data for better domain-specific performance. Prepare training data, manage training jobs, evaluate improvements, and deploy custom models.
Intelligent context window management for long conversations and documents. Summarization, sliding windows, and hierarchical context strategies to stay within token limits.
Implement function calling to let LLMs interact with your APIs and databases. Tool definitions, parameter validation, execution, and result integration back into conversations.
Route requests to appropriate models based on task complexity. Use GPT-4 for complex reasoning, GPT-3.5 for simple tasks, and specialized models for specific domains to optimize cost.
Extract, summarize, and analyze documents (PDFs, Word, HTML). Chunk long documents intelligently, extract structured data, generate summaries, and answer questions about content.
Build search that understands meaning, not just keywords. Generate embeddings with text-embedding models, implement hybrid search combining dense and sparse retrieval.
Automated content creation for blogs, product descriptions, emails, and social media. Template-based generation, style consistency, fact-checking, and quality control.
Reduce LLM costs by 60-80% through prompt optimization, response caching, model selection, request batching, and fallback to smaller models when appropriate.
Content filtering, bias detection, hallucination checks, and compliance with safety guidelines. Implement guardrails to prevent harmful or inaccurate outputs.
Define what the LLM needs to do, success criteria, and quality requirements. Identify data sources, user flows, and integration points.
Collect and organize your knowledge base, documents, FAQs, and domain-specific content. Prepare data for RAG, fine-tuning, or few-shot examples.
Design and test prompts for your use cases. Create system instructions, user message templates, and few-shot examples for consistent outputs.
Build vector database, implement chunking strategy, generate embeddings, and create retrieval pipeline. Optimize for accuracy and speed.
Integrate with LLM providers using best practices. Implement streaming, error handling, retries, fallbacks, and monitoring.
Create test cases covering edge cases. Evaluate accuracy, relevance, safety, and cost. A/B test different prompts and models.
Deploy to production with proper caching, rate limiting, and monitoring. Set up dashboards for tracking usage, costs, and quality metrics.
Continuously improve based on production data. Optimize prompts, update knowledge base, adjust retrieval, and reduce costs while maintaining quality.
Let's discuss your project and see how we can help bring your vision to life.
START A PROJECT# Production-Ready RAG System with GPT-4 and Vector Search
from openai import AsyncOpenAI
from typing import List, Dict, Optional
import tiktoken
from pinecone import Pinecone, ServerlessSpec
from sentence_transformers import SentenceTransformer
import asyncio
from functools import lru_cache
import hashlib
import redis
import json
class RAGSystem:
def __init__(self, openai_api_key: str, pinecone_api_key: str):
# Initialize OpenAI client
self.openai_client = AsyncOpenAI(api_key=openai_api_key)
self.model = "gpt-4-turbo-preview"
self.embedding_model_name = "text-embedding-3-large"
# Initialize Pinecone for vector storage
self.pc = Pinecone(api_key=pinecone_api_key)
self.index_name = "knowledge-base"
# Create index if it doesn't exist
if self.index_name not in self.pc.list_indexes().names():
self.pc.create_index(
name=self.index_name,
dimension=1536, # text-embedding-3-large dimension
metric="cosine",
spec=ServerlessSpec(cloud="aws", region="us-east-1")
)
self.index = self.pc.Index(self.index_name)
# Initialize embedding model for local encoding (faster for queries)
self.sentence_encoder = SentenceTransformer(
'sentence-transformers/all-MiniLM-L6-v2')
# Token counter
self.encoding = tiktoken.encoding_for_model(self.model)
# Redis for caching responses
self.redis_client = redis.Redis(
host='localhost', port=6379, decode_responses=True)
self.cache_ttl = 3600 # 1 hour
def count_tokens(self, text: str) -> int:
"""Count tokens in text"""
return len(self.encoding.encode(text))
async def embed_text(self, text: str) -> List[float]:
"""Generate embeddings using OpenAI"""
response = await self.openai_client.embeddings.create(
model=self.embedding_model_name,
input=text
)
return response.data[0].embedding
def chunk_document(self, text: str, chunk_size: int = 1000,
overlap: int = 200) -> List[Dict]:
"""
Split document into overlapping chunks for better retrieval
"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
# Try to break at sentence boundary
if end < len(text):
last_period = chunk.rfind('.')
last_newline = chunk.rfind('\n')
break_point = max(last_period, last_newline)
if break_point > chunk_size * 0.7: # At least 70% through
chunk = chunk[:break_point + 1]
end = start + break_point + 1
chunks.append({
'text': chunk.strip(),
'start': start,
'end': end
})
start = end - overlap
return chunks
async def index_document(self, document_id: str, text: str,
metadata: Optional[Dict] = None):
"""
Index a document by chunking and storing in vector database
"""
chunks = self.chunk_document(text)
# Generate embeddings for all chunks
vectors = []
for i, chunk in enumerate(chunks):
embedding = await self.embed_text(chunk['text'])
chunk_metadata = {
'document_id': document_id,
'chunk_index': i,
'text': chunk['text'],
'char_start': chunk['start'],
'char_end': chunk['end']
}
if metadata:
chunk_metadata.update(metadata)
vectors.append({
'id': f"{document_id}_chunk_{i}",
'values': embedding,
'metadata': chunk_metadata
})
# Upsert to Pinecone in batches
batch_size = 100
for i in range(0, len(vectors), batch_size):
batch = vectors[i:i + batch_size]
self.index.upsert(vectors=batch)
return len(chunks)
async def retrieve_context(self, query: str, top_k: int = 5,
filter_metadata: Optional[Dict] = None
) -> List[Dict]:
"""
Retrieve most relevant chunks for a query
"""
# Generate query embedding
query_embedding = await self.embed_text(query)
# Search vector database
results = self.index.query(
vector=query_embedding,
top_k=top_k,
include_metadata=True,
filter=filter_metadata
)
# Extract and re-rank results
contexts = []
for match in results.matches:
contexts.append({
'text': match.metadata['text'],
'score': match.score,
'document_id': match.metadata['document_id'],
'chunk_index': match.metadata['chunk_index']
})
return contexts
def _get_cache_key(self, query: str, context_texts: List[str]) -> str:
"""Generate cache key for query + context"""
content = query + "".join(context_texts)
return hashlib.sha256(content.encode()).hexdigest()
async def generate_answer(self, query: str,
contexts: List[Dict],
system_prompt: Optional[str] = None,
temperature: float = 0.7) -> Dict:
"""
Generate answer using LLM with retrieved context
"""
# Build context string
context_texts = [ctx['text'] for ctx in contexts]
context_str = "\n\n---\n\n".join(context_texts)
# Check cache
cache_key = self._get_cache_key(query, context_texts)
cached_response = self.redis_client.get(cache_key)
if cached_response:
return json.loads(cached_response)
# Build prompt
default_system_prompt = """You are a helpful assistant that answers
questions based on the provided context. If the answer cannot be found in
the context, say so clearly. Always cite which part of the context you used."""
system_message = system_prompt or default_system_prompt
messages = [
{"role": "system", "content": system_message},
{"role": "user", "content": f"""Context:
{context_str}
Question: {query}
Please answer the question based on the context provided above."""}
]
# Count tokens to estimate cost
prompt_tokens = sum(self.count_tokens(msg["content"])
for msg in messages)
# Generate response
response = await self.openai_client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=1000
)
answer = response.choices[0].message.content
completion_tokens = response.usage.completion_tokens
result = {
'answer': answer,
'contexts': contexts,
'prompt_tokens': prompt_tokens,
'completion_tokens': completion_tokens,
'total_tokens': prompt_tokens + completion_tokens,
'model': self.model
}
# Cache response
self.redis_client.setex(
cache_key,
self.cache_ttl,
json.dumps(result)
)
return result
async def ask(self, query: str, top_k: int = 5,
filter_metadata: Optional[Dict] = None) -> Dict:
"""
Main RAG pipeline: retrieve context and generate answer
"""
# Step 1: Retrieve relevant context
contexts = await self.retrieve_context(
query, top_k=top_k, filter_metadata=filter_metadata)
if not contexts:
return {
'answer': "I couldn't find relevant information to answer your question.",
'contexts': [],
'total_tokens': 0
}
# Step 2: Generate answer
result = await self.generate_answer(query, contexts)
return result
# Example usage
async def main():
rag = RAGSystem(
openai_api_key="your-openai-key",
pinecone_api_key="your-pinecone-key"
)
# Index documents
await rag.index_document(
document_id="doc_001",
text="""Your company's documentation, knowledge base articles,
product manuals, or any text content you want to query.""",
metadata={"source": "manual", "category": "product"}
)
# Query the system
result = await rag.ask(
"What are the key features of the product?",
top_k=3
)
print(f"Answer: {result['answer']}")
print(f"\nTokens used: {result['total_tokens']}")
print(f"\nSources used: {len(result['contexts'])} chunks")
if __name__ == "__main__":
asyncio.run(main())This production-ready RAG system includes document chunking, vector search with Pinecone,
response caching with Redis, token counting, and cost optimization strategies.