ai-engineer-expert

name	ai-engineer-expert
version	1.0.0
description	Expert-level AI implementation, deployment, LLM integration, and production AI systems
category	ai
tags	["ai-engineering","llm","deployment","production-ai","integration"]
allowed-tools	["Read","Write","Edit","Bash(python:*)"]

AI Engineer Expert

Expert guidance for implementing AI systems, LLM integration, prompt engineering, and deploying production AI applications.

Core Concepts

AI Engineering

• LLM integration and orchestration
• Prompt engineering and optimization
• RAG (Retrieval-Augmented Generation)
• Vector databases and embeddings
• Fine-tuning and adaptation
• AI agent systems

Production AI

• Model deployment strategies
• API design for AI services
• Rate limiting and cost control
• Error handling and fallbacks
• Monitoring and logging
• Security and safety

LLM Patterns

• Chain-of-thought prompting
• Few-shot learning
• System/user message design
• Function calling and tools
• Streaming responses
• Context window management

LLM Integration

from openai import AsyncOpenAI
from anthropic import Anthropic
from typing import List, Dict, Optional
import asyncio

class LLMClient:
    """Unified LLM client with fallback"""

    def __init__(self, primary: str = "openai", fallback: str = "anthropic"):
        self.openai_client = AsyncOpenAI()
        self.anthropic_client = Anthropic()
        self.primary = primary
        self.fallback = fallback

    async def chat_completion(self, messages: List[Dict],
                              model: str = "gpt-4-turbo",
                              temperature: float = 0.7,
                              max_tokens: int = 1000) -> str:
        """Chat completion with fallback"""
        try:
            if self.primary == "openai":
                response = await self.openai_client.chat.completions.create(
                    model=model,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=max_tokens
                )
                 response.choices[].message.content

         Exception  e:
            ()

             .fallback == :
                response = .anthropic_client.messages.create(
                    model=,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=max_tokens
                )
                 response.content[].text

      ():
        
        stream =  .openai_client.chat.completions.create(
            model=model,
            messages=messages,
            stream=
        )

          chunk  stream:
             chunk.choices[].delta.content:
                 chunk.choices[].delta.content

      () -> :
        
        response =  .openai_client.chat.completions.create(
            model=,
            messages=messages,
            tools=tools,
            tool_choice=
        )

        message = response.choices[].message

         message.tool_calls:
             {
                : ,
                : message.tool_calls[].function.name,
                : message.tool_calls[].function.arguments
            }
        :
             {
                : ,
                : message.content
            }

RAG Implementation

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

class RAGSystem:
    """Retrieval-Augmented Generation system"""

    def __init__(self, persist_directory: str = "./chroma_db"):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = None
        self.persist_directory = persist_directory
        self.llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

    def ingest_documents(self, documents: List[str]):
        """Ingest and index documents"""
        # Split documents into chunks
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        chunks = text_splitter.create_documents(documents)

        # Create vector store
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )

    def query(self, question: , k:  = ) -> :
        
          .vectorstore:
             ValueError()

        
        retriever = .vectorstore.as_retriever(
            search_kwargs={: k}
        )

        
        qa_chain = RetrievalQA.from_chain_type(
            llm=.llm,
            chain_type=,
            retriever=retriever,
            return_source_documents=
        )

        
        result = qa_chain({: question})

         {
            : result[],
            : [doc.page_content  doc  result[]]
        }

     () -> []:
        
        results = .vectorstore.similarity_search_with_score(query, k=k)

         [
            {
                : doc.page_content,
                : score,
                : doc.metadata
            }
             doc, score  results
        ]

Prompt Engineering

class PromptTemplate:
    """Advanced prompt templates"""

    @staticmethod
    def chain_of_thought(question: str) -> str:
        """Chain-of-thought prompting"""
        return f"""Let's solve this step by step:

Question: {question}

Please think through this problem carefully:
1. First, identify what we need to find
2. Then, break down the problem into smaller steps
3. Solve each step
4. Finally, combine the results

Your step-by-step solution:"""

    @staticmethod
    def few_shot(task: str, examples: List[Dict], query: str) -> str:
        """Few-shot learning prompt"""
        examples_text = "\n\n".join([
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in examples
        ])

        return f"""Task: {task}

Here are some examples:

{examples_text}

Now, please solve this:

Input: {query}
Output:"""

    @staticmethod
    def system_message(role: str, constraints: List[str],
                       format_instructions: str) -> str:
        """System message template"""
        constraints_text = .join([  c  constraints])

         

AI Agent System

from typing import Callable
import json

class Tool:
    """Tool that agents can use"""

    def __init__(self, name: str, description: str, function: Callable):
        self.name = name
        self.description = description
        self.function = function

    def to_openai_function(self) -> Dict:
        """Convert to OpenAI function format"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.get_parameters()
            }
        }

class AIAgent:
    """AI agent with tools"""

    def __init__(self, llm_client: LLMClient, tools: List[Tool]):
        self.llm = llm_client
        self.tools = {tool.name: tool for tool in tools}
        self.conversation_history = []

    async def run() -> :
        
        .conversation_history.append({
            : ,
            : user_input
        })

         i  (max_iterations):
            
            response =  .llm.function_calling(
                messages=.conversation_history,
                tools=[tool.to_openai_function()  tool  .tools.values()]
            )

             response[] == :
                
                 response[]

            
            tool_name = response[]
            arguments = json.loads(response[])

            tool_result =  .execute_tool(tool_name, arguments)

            
            .conversation_history.append({
                : ,
                : tool_name,
                : (tool_result)
            })

         

      () -> :
        
         tool_name   .tools:
             ValueError()

        tool = .tools[tool_name]
          tool.function(**arguments)

Production Deployment

from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from circuitbreaker import circuit
import asyncio

app = FastAPI()

class ChatRequest(BaseModel):
    messages: List[Dict]
    model: str = "gpt-4-turbo"
    stream: bool = False

class RateLimiter:
    """Rate limiter for API"""

    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = {}

    async def check_limit(self, user_id: str) -> bool:
        """Check if user is within rate limit"""
        import time
        now = time.time()

        if user_id not in self.requests:
            self.requests[user_id] = []

        # Remove old requests
        self.requests[user_id] = [
            req_time for req_time in .requests[user_id]
             now - req_time < .window_seconds
        ]

         (.requests[user_id]) >= .max_requests:
             

        .requests[user_id].append(now)
         

rate_limiter = RateLimiter(max_requests=, window_seconds=)
llm_client = LLMClient()


  () -> :
    
      llm_client.chat_completion(messages)


  ():
    
    
       rate_limiter.check_limit(user_id):
         HTTPException(status_code=, detail=)

    :
         request.stream:
              ():
                  chunk  llm_client.chat_completion_streaming(request.messages):
                     chunk

             StreamingResponse(generate(), media_type=)
        :
            response =  call_llm(request.messages)
             {: response}

     Exception  e:
         HTTPException(status_code=, detail=(e))

Best Practices

LLM Integration

• Implement fallback providers
• Use streaming for better UX
• Cache responses where appropriate
• Handle rate limits gracefully
• Monitor token usage and costs
• Version prompts and track changes

Production Systems

• Implement circuit breakers
• Add comprehensive logging
• Monitor latency and errors
• Use rate limiting
• Implement retry logic with backoff
• Test edge cases thoroughly

Security

• Validate and sanitize inputs
• Implement authentication/authorization
• Never expose API keys in logs
• Use environment variables for secrets
• Implement content filtering
• Monitor for prompt injection

Anti-Patterns

❌ No error handling or fallbacks ❌ Exposing raw LLM outputs without validation ❌ No rate limiting or cost controls ❌ Storing API keys in code ❌ No monitoring or logging ❌ Ignoring token limits ❌ No testing of prompts

Resources

• OpenAI API: https://platform.openai.com/docs
• Anthropic Claude: https://docs.anthropic.com/
• LangChain: https://python.langchain.com/
• LlamaIndex: https://www.llamaindex.ai/
• Weights & Biases Prompts: https://wandb.ai/site/prompts