Vertex AI Platform Fundamentals¶

from google.cloud import aiplatform

# Initialize Vertex AI
aiplatform.init(project='my-project', location='us-central1')

# Create custom training job
job = aiplatform.CustomTrainingJob(
    display_name='my-training-job',
    script_path='train.py',
    container_uri='gcr.io/cloud-aiplatform/training/pytorch-gpu.1-9:latest',
    requirements=['scikit-learn', 'pandas'],
    model_serving_container_image_uri='gcr.io/cloud-aiplatform/prediction/pytorch-gpu.1-9:latest'
)

# Run training
model = job.run(
    dataset=my_dataset,
    replica_count=1,
    machine_type='n1-standard-4',
    accelerator_type='NVIDIA_TESLA_K80',
    accelerator_count=1
)

# Deploy model to endpoint
endpoint = model.deploy(
    machine_type='n1-standard-2',
    min_replica_count=1,
    max_replica_count=10,
    accelerator_type='NVIDIA_TESLA_T4',
    accelerator_count=1
)

# Make predictions
predictions = endpoint.predict(instances=test_data)

from kfp.v2 import dsl
from kfp.v2.dsl import component, pipeline, Input, Output, Dataset, Model

@component(
    base_image='python:3.9',
    packages_to_install=['pandas', 'scikit-learn']
)
def preprocess_data(
    input_data: Input[Dataset],
    output_data: Output[Dataset]
):
    import pandas as pd

    # Load and preprocess data
    df = pd.read_csv(input_data.path)
    # ... preprocessing logic
    df.to_csv(output_data.path, index=False)

@component(
    base_image='python:3.9',
    packages_to_install=['scikit-learn']
)
def train_model(
    training_data: Input[Dataset],
    model: Output[Model]
):
    import pickle
    from sklearn.ensemble import RandomForestClassifier

    # Train model
    # ... training logic

    # Save model
    with open(model.path, 'wb') as f:
        pickle.dump(trained_model, f)

@pipeline(name='ml-pipeline')
def ml_pipeline():
    preprocess_task = preprocess_data(input_data=dataset)
    train_task = train_model(training_data=preprocess_task.outputs['output_data'])

# Register model
model = aiplatform.Model.upload(
    display_name='my-model',
    artifact_uri='gs://my-bucket/model/',
    serving_container_image_uri='gcr.io/cloud-aiplatform/prediction/sklearn-cpu.0-24:latest',
    description='My custom model'
)

# Create model version
model_version = model.upload(
    parent_model=model,
    version_description='Updated model with new features'
)

import vertexai
from vertexai.language_models import TextGenerationModel

# Initialize Vertex AI
vertexai.init(project="my-project", location="us-central1")

# Load text-bison model
model = TextGenerationModel.from_pretrained("text-bison@002")

# Basic text generation
response = model.predict(
    prompt="Write a comprehensive product description for a smartwatch with health monitoring features",
    temperature=0.7,          # Controls randomness (0.0-1.0)
    max_output_tokens=1024,   # Maximum response length
    top_k=40,                 # Top-k sampling
    top_p=0.95,               # Nucleus sampling
    stop_sequences=["\n\n"]   # Stop generation at sequences
)

print(f"Response: {response.text}")
print(f"Safety attributes: {response.safety_attributes}")

from vertexai.language_models import ChatModel

# Load chat model
chat_model = ChatModel.from_pretrained("chat-bison@002")

# Start chat session
chat = chat_model.start_chat(
    context="You are a helpful cloud architect assistant specializing in GCP",
    examples=[
        {
            "input": {"content": "What is Vertex AI?"},
            "output": {"content": "Vertex AI is Google Cloud's unified ML platform..."}
        }
    ],
    temperature=0.3,
    max_output_tokens=256,
    top_k=40,
    top_p=0.8
)

# Send messages
response1 = chat.send_message("What are the benefits of using Vertex AI?")
print(f"Response 1: {response1.text}")

# Follow-up with context
response2 = chat.send_message("How does it compare to AWS SageMaker?")
print(f"Response 2: {response2.text}")

# Get full conversation history
print(f"Message history: {chat.message_history}")

from vertexai.language_models import CodeGenerationModel

# Load code generation model
code_model = CodeGenerationModel.from_pretrained("code-bison@002")

# Generate code
response = code_model.predict(
    prefix="""
# Write a Python function that implements a retry decorator
# with exponential backoff for API calls
""",
    temperature=0.2,  # Lower temperature for more deterministic code
    max_output_tokens=512
)

print(f"Generated code:\n{response.text}")

# Code completion
completion = code_model.predict(
    prefix="""
def calculate_fibonacci(n):
    \"\"\"Calculate fibonacci number using dynamic programming\"\"\"
    if n <= 1:
        return n

    # Complete this function
""",
    suffix="""
    return fib[n]
""",
    temperature=0.1,
    max_output_tokens=256
)

print(f"Completed code:\n{completion.text}")

from vertexai.language_models import CodeChatModel

# Load code chat model
code_chat_model = CodeChatModel.from_pretrained("codechat-bison@002")

# Start code chat
code_chat = code_chat_model.start_chat()

# Ask about code
response1 = code_chat.send_message(
    "Write a Python class for a binary search tree with insert and search methods"
)
print(f"Code: {response1.text}")

# Follow up for modifications
response2 = code_chat.send_message(
    "Now add a delete method to this class"
)
print(f"Updated code: {response2.text}")

# Request explanations
response3 = code_chat.send_message(
    "Explain the time complexity of these operations"
)
print(f"Explanation: {response3.text}")

from vertexai.language_models import TextEmbeddingModel

# Load embedding model
embedding_model = TextEmbeddingModel.from_pretrained("textembedding-gecko@003")

# Generate single embedding
embedding = embedding_model.get_embeddings(["Cloud computing platforms"])[0]
print(f"Embedding vector (length {len(embedding.values)}): {embedding.values[:5]}...")

# Batch embeddings
texts = [
    "Vertex AI is a machine learning platform",
    "BigQuery is a data warehouse",
    "Cloud Storage stores objects",
    "Kubernetes runs containers"
]

embeddings = embedding_model.get_embeddings(texts)
for i, emb in enumerate(embeddings):
    print(f"Text {i}: {len(emb.values)} dimensions")

# Calculate similarity
import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Compare embeddings
emb1 = embedding_model.get_embeddings(["Machine learning"])[0].values
emb2 = embedding_model.get_embeddings(["Artificial intelligence"])[0].values
emb3 = embedding_model.get_embeddings(["Database systems"])[0].values

similarity_ml_ai = cosine_similarity(emb1, emb2)
similarity_ml_db = cosine_similarity(emb1, emb3)

print(f"ML vs AI similarity: {similarity_ml_ai:.4f}")
print(f"ML vs DB similarity: {similarity_ml_db:.4f}")

import vertexai
from vertexai.generative_models import GenerativeModel, Part, FinishReason
import vertexai.preview.generative_models as generative_models

# Initialize
vertexai.init(project="my-project", location="us-central1")

# Load Gemini Pro
model = GenerativeModel("gemini-1.0-pro")

# Basic generation
response = model.generate_content(
    "Explain the architecture of a microservices-based e-commerce platform",
    generation_config={
        "max_output_tokens": 2048,
        "temperature": 0.7,
        "top_p": 0.95,
        "top_k": 40
    },
    safety_settings={
        generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
        generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
        generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
        generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    }
)

print(f"Response: {response.text}")

# Multi-turn conversation
chat = model.start_chat()

response1 = chat.send_message("What is Kubernetes?")
print(f"Response 1: {response1.text}")

response2 = chat.send_message("How do I deploy a workload to it?")
print(f"Response 2: {response2.text}")

response3 = chat.send_message("What about scaling?")
print(f"Response 3: {response3.text}")

# Streaming responses
responses = model.generate_content(
    "Write a detailed guide to setting up a CI/CD pipeline",
    stream=True
)

print("Streaming response:")
for response in responses:
    print(response.text, end="")

from vertexai.generative_models import GenerativeModel, Part
import base64

# Load Gemini Pro Vision
vision_model = GenerativeModel("gemini-1.0-pro-vision")

# Image from Cloud Storage
image_part = Part.from_uri(
    uri="gs://my-bucket/architecture-diagram.png",
    mime_type="image/png"
)

response = vision_model.generate_content([
    "Analyze this architecture diagram and explain the components and data flow",
    image_part
])

print(f"Analysis: {response.text}")

# Multiple images
image1 = Part.from_uri("gs://my-bucket/before.jpg", mime_type="image/jpeg")
image2 = Part.from_uri("gs://my-bucket/after.jpg", mime_type="image/jpeg")

response = vision_model.generate_content([
    "Compare these two images and describe the differences:",
    image1,
    image2
])

print(f"Comparison: {response.text}")

# Image from local file
def load_image_from_file(file_path):
    with open(file_path, "rb") as f:
        image_data = f.read()
    return Part.from_data(data=image_data, mime_type="image/jpeg")

local_image = load_image_from_file("/path/to/screenshot.jpg")
response = vision_model.generate_content([
    "Extract and format all text from this screenshot",
    local_image
])

print(f"Extracted text: {response.text}")

# Video understanding
video_part = Part.from_uri(
    uri="gs://my-bucket/tutorial-video.mp4",
    mime_type="video/mp4"
)

response = vision_model.generate_content([
    "Summarize the key points from this tutorial video",
    video_part
])

print(f"Summary: {response.text}")

from vertexai.generative_models import (
    GenerativeModel,
    Tool,
    FunctionDeclaration,
    Part
)
import json

# Define function declarations
get_weather_func = FunctionDeclaration(
    name="get_current_weather",
    description="Get the current weather in a given location",
    parameters={
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "The city and state, e.g. San Francisco, CA"
            },
            "unit": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"],
                "description": "Temperature unit"
            }
        },
        "required": ["location"]
    }
)

search_product_func = FunctionDeclaration(
    name="search_products",
    description="Search for products in inventory",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search query"
            },
            "category": {
                "type": "string",
                "description": "Product category"
            },
            "max_results": {
                "type": "integer",
                "description": "Maximum number of results to return"
            }
        },
        "required": ["query"]
    }
)

# Create tool
tools = Tool(
    function_declarations=[get_weather_func, search_product_func]
)

# Initialize model with tools
model = GenerativeModel(
    "gemini-1.0-pro",
    tools=[tools]
)

# Send request
chat = model.start_chat()
response = chat.send_message("What's the weather like in Boston?")

# Extract function call
function_call = response.candidates[0].content.parts[0].function_call
print(f"Function: {function_call.name}")
print(f"Arguments: {dict(function_call.args)}")

# Simulate function execution
def get_current_weather(location, unit="fahrenheit"):
    # Simulate API call
    return {
        "location": location,
        "temperature": 72,
        "unit": unit,
        "conditions": "Partly cloudy"
    }

# Execute function
weather_data = get_current_weather(
    location=function_call.args["location"],
    unit=function_call.args.get("unit", "fahrenheit")
)

# Return result to model
response = chat.send_message(
    Part.from_function_response(
        name="get_current_weather",
        response={"content": weather_data}
    )
)

print(f"Final response: {response.text}")

# JSON mode for structured output
model = GenerativeModel("gemini-1.0-pro")

prompt = """
Extract the following information from this text and return as JSON:
- Company name
- Products mentioned
- Key features
- Pricing

Text: "Acme Corp offers CloudSync Pro for $99/month with features including
real-time synchronization, end-to-end encryption, and 1TB storage."
"""

response = model.generate_content(
    prompt,
    generation_config={
        "response_mime_type": "application/json"
    }
)

data = json.loads(response.text)
print(f"Structured data: {json.dumps(data, indent=2)}")

from vertexai.language_models import TextGenerationModel
import pandas as pd

# Prepare training data
training_data = pd.DataFrame({
    "input_text": [
        "Classify the urgency: Customer cannot access account",
        "Classify the urgency: Request for product brochure",
        "Classify the urgency: System is completely down",
        # ... more examples
    ],
    "output_text": [
        "HIGH: Account access issue requires immediate attention",
        "LOW: Informational request, can be handled in regular queue",
        "CRITICAL: System outage affecting all users",
        # ... corresponding outputs
    ]
})

# Save to JSONL format
training_data.to_json("training_data.jsonl", orient="records", lines=True)

# Upload to Cloud Storage
from google.cloud import storage

storage_client = storage.Client()
bucket = storage_client.bucket("my-training-bucket")
blob = bucket.blob("tuning-data/training_data.jsonl")
blob.upload_from_filename("training_data.jsonl")

# Load base model
base_model = TextGenerationModel.from_pretrained("text-bison@002")

# Start tuning job
tuning_job = base_model.tune_model(
    training_data="gs://my-training-bucket/tuning-data/training_data.jsonl",
    train_steps=100,
    tuning_job_location="us-central1",
    tuned_model_location="us-central1",
    model_display_name="urgency-classifier-v1",
    tuning_evaluation_spec={
        "evaluation_data": "gs://my-training-bucket/tuning-data/eval_data.jsonl",
        "evaluation_interval": 20,
        "enable_early_stopping": True
    },
    learning_rate=0.001,
    learning_rate_multiplier=1.0,
)

print(f"Tuning job: {tuning_job.name}")
print(f"Status: {tuning_job.state}")

# Monitor tuning progress
tuning_job.wait()

# Get tuned model
tuned_model = tuning_job.get_tuned_model()

# Use tuned model
response = tuned_model.predict(
    prompt="Classify the urgency: Email server responding slowly",
    temperature=0.2,
    max_output_tokens=100
)

print(f"Classification: {response.text}")

# Compare with base model
base_response = base_model.predict(
    prompt="Classify the urgency: Email server responding slowly",
    temperature=0.2,
    max_output_tokens=100
)

print(f"Base model: {base_response.text}")
print(f"Tuned model: {response.text}")

from vertexai.language_models import ChatModel

# Prepare chat training data with context and examples
training_data = [
    {
        "input_text": "What are your business hours?",
        "output_text": "Our support team is available Monday through Friday, 9 AM to 5 PM EST. For urgent issues outside business hours, please use our emergency hotline at 1-800-SUPPORT."
    },
    {
        "input_text": "How do I reset my password?",
        "output_text": "To reset your password: 1) Go to the login page, 2) Click 'Forgot Password', 3) Enter your email, 4) Follow the link in your email, 5) Create a new password with at least 12 characters."
    },
    # ... more examples
]

# Save training data
import json
with open("chat_training.jsonl", "w") as f:
    for item in training_data:
        f.write(json.dumps(item) + "\n")

# Upload to GCS
blob = bucket.blob("tuning-data/chat_training.jsonl")
blob.upload_from_filename("chat_training.jsonl")

# Load base chat model
base_chat_model = ChatModel.from_pretrained("chat-bison@002")

# Tune chat model
tuning_job = base_chat_model.tune_model(
    training_data="gs://my-training-bucket/tuning-data/chat_training.jsonl",
    train_steps=200,
    tuning_job_location="us-central1",
    tuned_model_location="us-central1",
    model_display_name="customer-support-chat-v1",
    context="You are a helpful customer support agent for TechCorp. Be professional, concise, and provide step-by-step instructions when appropriate.",
    learning_rate_multiplier=1.0,
    tuning_evaluation_spec={
        "evaluation_data": "gs://my-training-bucket/tuning-data/chat_eval.jsonl",
        "evaluation_interval": 50
    }
)

# Wait for completion
tuning_job.wait()

# Use tuned chat model
tuned_chat_model = tuning_job.get_tuned_model()
chat = tuned_chat_model.start_chat()

response = chat.send_message("I can't log into my account")
print(f"Tuned model response: {response.text}")

# Adapter tuning for text model
adapter_tuning_job = base_model.tune_model(
    training_data="gs://my-training-bucket/tuning-data/training_data.jsonl",
    train_steps=50,  # Fewer steps needed
    adapter_size=4,  # Adapter dimensions (1, 4, 8, 16)
    tuning_job_location="us-central1",
    model_display_name="urgency-classifier-adapter-v1"
)

# Multiple adapters for different tasks
summarization_adapter = base_model.tune_model(
    training_data="gs://my-bucket/summarization_data.jsonl",
    train_steps=50,
    adapter_size=8,
    model_display_name="summarization-adapter"
)

classification_adapter = base_model.tune_model(
    training_data="gs://my-bucket/classification_data.jsonl",
    train_steps=50,
    adapter_size=8,
    model_display_name="classification-adapter"
)

# Switch between adapters at inference time
summary = summarization_adapter.get_tuned_model().predict("Long text to summarize...")
category = classification_adapter.get_tuned_model().predict("Text to classify...")

# Note: RLHF typically requires custom implementation
# This shows the conceptual approach

# Step 1: Collect preference data
preference_data = [
    {
        "prompt": "Explain quantum computing",
        "response_a": "Response A text...",
        "response_b": "Response B text...",
        "preferred": "response_b",  # Human preference
        "feedback": "More clear and concise"
    },
    # ... more preference pairs
]

# Step 2: Prepare RLHF training data
# Convert preferences to reward signals
rlhf_training_data = []
for item in preference_data:
    rlhf_training_data.append({
        "input_text": item["prompt"],
        "output_text": item[item["preferred"]],
        "reward": 1.0
    })

    # Add negative example
    not_preferred = "response_a" if item["preferred"] == "response_b" else "response_b"
    rlhf_training_data.append({
        "input_text": item["prompt"],
        "output_text": item[not_preferred],
        "reward": -0.5
    })

# Step 3: Tune with preference data
# This uses standard tuning but with preference-optimized data
tuned_model = base_model.tune_model(
    training_data="gs://my-bucket/rlhf_training.jsonl",
    train_steps=100,
    learning_rate_multiplier=0.5  # Lower LR for stability
)

from vertexai.preview.generative_models import (
    GenerativeModel,
    Tool,
    grounding
)

# Initialize model with Google Search grounding
model = GenerativeModel("gemini-1.0-pro")

# Create grounding tool
google_search_tool = Tool.from_google_search_retrieval(
    grounding.GoogleSearchRetrieval()
)

# Generate with grounding
response = model.generate_content(
    "What are the latest developments in quantum computing announced in 2024?",
    tools=[google_search_tool],
    generation_config={
        "temperature": 0.3,
        "max_output_tokens": 1024
    }
)

print(f"Response: {response.text}")

# Access grounding metadata
if response.candidates[0].grounding_metadata:
    grounding_metadata = response.candidates[0].grounding_metadata
    print(f"\nGrounding sources:")
    for chunk in grounding_metadata.grounding_chunks:
        print(f"  - {chunk.web.uri}")
        print(f"    Title: {chunk.web.title}")

    print(f"\nGrounding support: {grounding_metadata.grounding_support}")

from vertexai.preview.generative_models import grounding

# Create Vertex AI Search datastore
# (Done via Console or gcloud CLI)
# gcloud alpha discovery-engine data-stores create my-datastore \
#   --location=global \
#   --collection=default_collection \
#   --industry-vertical=GENERIC

# Initialize model with Vertex AI Search grounding
vertex_search_tool = Tool.from_retrieval(
    grounding.Retrieval(
        source=grounding.VertexAISearch(
            datastore="projects/my-project/locations/global/collections/default_collection/dataStores/my-datastore"
        )
    )
)

model = GenerativeModel("gemini-1.0-pro")

# Query with grounded responses
response = model.generate_content(
    "What is our company's return policy for electronics?",
    tools=[vertex_search_tool]
)

print(f"Grounded response: {response.text}")

# Get source documents
for chunk in response.candidates[0].grounding_metadata.grounding_chunks:
    print(f"Source: {chunk.retrieved_context.uri}")
    print(f"Content: {chunk.retrieved_context.text[:200]}...")

# Grounding using Vector Search / Matching Engine
from google.cloud import aiplatform_v1
from vertexai.preview.generative_models import GenerativeModel
import numpy as np

# Step 1: Create vector embeddings of your documents
from vertexai.language_models import TextEmbeddingModel

embedding_model = TextEmbeddingModel.from_pretrained("textembedding-gecko@003")

documents = [
    {
        "id": "doc1",
        "text": "Our return policy allows returns within 30 days...",
        "category": "policy"
    },
    {
        "id": "doc2",
        "text": "Product warranty covers manufacturing defects for 1 year...",
        "category": "warranty"
    },
    # ... more documents
]

# Generate embeddings
for doc in documents:
    embedding = embedding_model.get_embeddings([doc["text"]])[0]
    doc["embedding"] = embedding.values

# Step 2: Store in Vector Search (Matching Engine)
# Create index (done via API or Console)

# Step 3: Query-time retrieval
def retrieve_relevant_docs(query, top_k=5):
    # Get query embedding
    query_embedding = embedding_model.get_embeddings([query])[0].values

    # Search vector index
    # (Simplified - actual implementation uses Matching Engine API)
    similarities = []
    for doc in documents:
        similarity = np.dot(query_embedding, doc["embedding"]) / (
            np.linalg.norm(query_embedding) * np.linalg.norm(doc["embedding"])
        )
        similarities.append((similarity, doc))

    # Return top-k most similar
    similarities.sort(reverse=True)
    return [doc for _, doc in similarities[:top_k]]

# Step 4: Use retrieved docs as context
query = "What is the return policy?"
relevant_docs = retrieve_relevant_docs(query)

# Build context from retrieved documents
context = "\n\n".join([
    f"[Source {i+1}]\n{doc['text']}"
    for i, doc in enumerate(relevant_docs)
])

# Generate with context
model = GenerativeModel("gemini-1.0-pro")
prompt = f"""Based on the following information, answer the question.

Context:
{context}

Question: {query}

Answer:"""

response = model.generate_content(prompt)
print(f"Answer: {response.text}")
print(f"\nSources used:")
for i, doc in enumerate(relevant_docs):
    print(f"  {i+1}. {doc['id']}: {doc['text'][:100]}...")

from vertexai.generative_models import (
    GenerativeModel,
    Tool,
    FunctionDeclaration,
    Part
)
import requests

# Define function to call external API
get_stock_price_func = FunctionDeclaration(
    name="get_stock_price",
    description="Get current stock price for a ticker symbol",
    parameters={
        "type": "object",
        "properties": {
            "ticker": {
                "type": "string",
                "description": "Stock ticker symbol (e.g., GOOGL, AAPL)"
            }
        },
        "required": ["ticker"]
    }
)

get_exchange_rate_func = FunctionDeclaration(
    name="get_exchange_rate",
    description="Get currency exchange rate",
    parameters={
        "type": "object",
        "properties": {
            "from_currency": {
                "type": "string",
                "description": "Source currency code (e.g., USD)"
            },
            "to_currency": {
                "type": "string",
                "description": "Target currency code (e.g., EUR)"
            }
        },
        "required": ["from_currency", "to_currency"]
    }
)

# Create tool
financial_tool = Tool(
    function_declarations=[get_stock_price_func, get_exchange_rate_func]
)

# Initialize model
model = GenerativeModel("gemini-1.0-pro", tools=[financial_tool])

# Implementation of functions
def get_stock_price(ticker):
    # Call actual financial API
    # response = requests.get(f"https://api.example.com/stock/{ticker}")
    # return response.json()

    # Simulated response
    return {
        "ticker": ticker,
        "price": 142.50,
        "currency": "USD",
        "timestamp": "2024-01-15T10:30:00Z"
    }

def get_exchange_rate(from_currency, to_currency):
    # Call actual exchange rate API
    return {
        "from": from_currency,
        "to": to_currency,
        "rate": 0.85,
        "timestamp": "2024-01-15T10:30:00Z"
    }

# Use model with function calling
chat = model.start_chat()

response = chat.send_message(
    "What is the current price of Google stock and convert it to Euros?"
)

# Process function calls
while response.candidates[0].content.parts[0].function_call:
    function_call = response.candidates[0].content.parts[0].function_call

    # Execute appropriate function
    if function_call.name == "get_stock_price":
        result = get_stock_price(function_call.args["ticker"])
    elif function_call.name == "get_exchange_rate":
        result = get_exchange_rate(
            function_call.args["from_currency"],
            function_call.args["to_currency"]
        )

    # Send result back to model
    response = chat.send_message(
        Part.from_function_response(
            name=function_call.name,
            response={"content": result}
        )
    )

# Final natural language response
print(f"Answer: {response.text}")

# Define multiple tools for complex workflows
database_query_func = FunctionDeclaration(
    name="query_database",
    description="Query the customer database",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "SQL query to execute"},
            "database": {"type": "string", "description": "Database name"}
        },
        "required": ["query", "database"]
    }
)

send_email_func = FunctionDeclaration(
    name="send_email",
    description="Send an email to a customer",
    parameters={
        "type": "object",
        "properties": {
            "to": {"type": "string", "description": "Recipient email"},
            "subject": {"type": "string", "description": "Email subject"},
            "body": {"type": "string", "description": "Email body"}
        },
        "required": ["to", "subject", "body"]
    }
)

create_ticket_func = FunctionDeclaration(
    name="create_support_ticket",
    description="Create a customer support ticket",
    parameters={
        "type": "object",
        "properties": {
            "customer_id": {"type": "string"},
            "priority": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
            "description": {"type": "string"}
        },
        "required": ["customer_id", "description"]
    }
)

# Create tool with multiple functions
support_tool = Tool(
    function_declarations=[database_query_func, send_email_func, create_ticket_func]
)

# Initialize model
model = GenerativeModel("gemini-1.0-pro", tools=[support_tool])

# Complex multi-step request
chat = model.start_chat()
response = chat.send_message(
    """
    Customer John Doe (john@example.com) reported that his order #12345 hasn't arrived.
    Please:
    1. Look up the order status in the database
    2. Create a high-priority support ticket
    3. Send an email to the customer acknowledging the issue
    """
)

# Process multiple function calls
function_responses = []
while True:
    if not response.candidates[0].content.parts[0].function_call:
        break

    function_call = response.candidates[0].content.parts[0].function_call
    print(f"Executing: {function_call.name}")

    # Execute function based on name
    if function_call.name == "query_database":
        result = query_database(function_call.args["query"], function_call.args["database"])
    elif function_call.name == "create_support_ticket":
        result = create_support_ticket(**dict(function_call.args))
    elif function_call.name == "send_email":
        result = send_email(**dict(function_call.args))

    # Send result back
    response = chat.send_message(
        Part.from_function_response(
            name=function_call.name,
            response={"content": result}
        )
    )

    function_responses.append({
        "function": function_call.name,
        "result": result
    })

print(f"Final response: {response.text}")
print(f"Functions executed: {len(function_responses)}")

from google.cloud import storage, documentai_v1
from vertexai.language_models import TextEmbeddingModel
import PyPDF2
import json

# Initialize services
storage_client = storage.Client()
embedding_model = TextEmbeddingModel.from_pretrained("textembedding-gecko@003")

# Step 1: Extract text from documents
def extract_text_from_pdf(gcs_uri):
    """Extract text from PDF in Cloud Storage"""
    bucket_name, blob_name = gcs_uri.replace("gs://", "").split("/", 1)
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(blob_name)

    # Download PDF
    pdf_bytes = blob.download_as_bytes()

    # Extract text
    pdf_reader = PyPDF2.PdfReader(io.BytesIO(pdf_bytes))
    text = ""
    for page in pdf_reader.pages:
        text += page.extract_text()

    return text

# Step 2: Chunk documents
def chunk_text(text, chunk_size=500, overlap=50):
    """Split text into overlapping chunks"""
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

# Step 3: Generate embeddings
def embed_chunks(chunks, metadata=None):
    """Generate embeddings for text chunks"""
    embedded_chunks = []

    # Batch process for efficiency
    batch_size = 250
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        embeddings = embedding_model.get_embeddings(batch)

        for j, embedding in enumerate(embeddings):
            embedded_chunks.append({
                "text": batch[j],
                "embedding": embedding.values,
                "metadata": metadata[i + j] if metadata else {}
            })

    return embedded_chunks

# Process document
document_uri = "gs://my-bucket/documents/product-manual.pdf"
text = extract_text_from_pdf(document_uri)
chunks = chunk_text(text, chunk_size=500, overlap=50)

metadata = [
    {
        "source": document_uri,
        "chunk_id": i,
        "document_type": "product_manual"
    }
    for i in range(len(chunks))
]

embedded_chunks = embed_chunks(chunks, metadata)
print(f"Processed {len(embedded_chunks)} chunks")

from google.cloud import aiplatform
import json

# Initialize Vertex AI
aiplatform.init(project="my-project", location="us-central1")

# Create Matching Engine Index
def create_vector_index(embedded_chunks, index_name):
    """Create vector search index"""

    # Prepare data for index
    vectors = []
    for i, chunk in enumerate(embedded_chunks):
        vectors.append({
            "id": str(i),
            "embedding": chunk["embedding"],
            "restricts": [
                {"namespace": "document_type", "allow_list": [chunk["metadata"]["document_type"]]}
            ]
        })

    # Save vectors to GCS
    vectors_file = f"gs://my-bucket/vector-index/vectors.json"
    with open("vectors.json", "w") as f:
        for vector in vectors:
            f.write(json.dumps(vector) + "\n")

    # Upload to GCS
    bucket = storage_client.bucket("my-bucket")
    blob = bucket.blob("vector-index/vectors.json")
    blob.upload_from_filename("vectors.json")

    # Create index
    index = aiplatform.MatchingEngineIndex.create_tree_ah_index(
        display_name=index_name,
        contents_delta_uri=vectors_file,
        dimensions=768,  # textembedding-gecko dimensions
        approximate_neighbors_count=10,
        distance_measure_type="DOT_PRODUCT_DISTANCE",
        leaf_node_embedding_count=500,
        leaf_nodes_to_search_percent=5,
    )

    return index

# Create index
# index = create_vector_index(embedded_chunks, "document-index-v1")

# Deploy index to endpoint
def deploy_index(index):
    """Deploy index to endpoint"""
    endpoint = aiplatform.MatchingEngineIndexEndpoint.create(
        display_name="document-search-endpoint",
        public_endpoint_enabled=True
    )

    endpoint.deploy_index(
        index=index,
        deployed_index_id="doc_index_deployed",
        display_name="Document Search",
        machine_type="n1-standard-2",
        min_replica_count=1,
        max_replica_count=5
    )

    return endpoint

# endpoint = deploy_index(index)

def search_similar_chunks(query, endpoint, top_k=5):
    """Search for similar chunks using vector similarity"""

    # Generate query embedding
    query_embedding = embedding_model.get_embeddings([query])[0].values

    # Query index endpoint
    response = endpoint.find_neighbors(
        deployed_index_id="doc_index_deployed",
        queries=[query_embedding],
        num_neighbors=top_k
    )

    # Get matching chunks
    matches = []
    for neighbor in response[0]:
        chunk_id = int(neighbor.id)
        chunk = embedded_chunks[chunk_id]
        matches.append({
            "text": chunk["text"],
            "metadata": chunk["metadata"],
            "score": neighbor.distance
        })

    return matches

# Search for relevant information
query = "How do I troubleshoot connectivity issues?"
relevant_chunks = search_similar_chunks(query, endpoint, top_k=5)

for i, chunk in enumerate(relevant_chunks):
    print(f"\n{i+1}. Score: {chunk['score']:.4f}")
    print(f"   Text: {chunk['text'][:200]}...")
    print(f"   Source: {chunk['metadata']['source']}")

from vertexai.generative_models import GenerativeModel

def rag_query(question, endpoint, model_name="gemini-1.0-pro"):
    """RAG: Retrieve relevant context and generate answer"""

    # Step 1: Retrieve relevant chunks
    relevant_chunks = search_similar_chunks(question, endpoint, top_k=5)

    # Step 2: Build context
    context = "\n\n".join([
        f"[Source {i+1}: {chunk['metadata']['source']}]\n{chunk['text']}"
        for i, chunk in enumerate(relevant_chunks)
    ])

    # Step 3: Generate with context
    model = GenerativeModel(model_name)

    prompt = f"""Answer the question based on the provided context. If the answer is not in the context, say "I don't have enough information to answer that."

Context:
{context}

Question: {question}

Answer:"""

    response = model.generate_content(
        prompt,
        generation_config={
            "temperature": 0.2,
            "max_output_tokens": 1024,
            "top_p": 0.8
        }
    )

    return {
        "answer": response.text,
        "sources": [chunk["metadata"] for chunk in relevant_chunks],
        "context_chunks": [chunk["text"] for chunk in relevant_chunks]
    }

# Use RAG system
result = rag_query("How do I troubleshoot connectivity issues?", endpoint)

print(f"Answer: {result['answer']}\n")
print(f"Sources:")
for i, source in enumerate(result['sources']):
    print(f"  {i+1}. {source['source']} (chunk {source['chunk_id']})")

def hybrid_search(query, endpoint, keyword_index, top_k=10):
    """Combine vector search with keyword search"""

    # Vector search
    vector_results = search_similar_chunks(query, endpoint, top_k=top_k)

    # Keyword search (using BigQuery or Elasticsearch)
    # Simplified example
    keyword_results = []
    query_words = query.lower().split()
    for chunk in embedded_chunks:
        chunk_words = chunk["text"].lower().split()
        overlap = len(set(query_words) & set(chunk_words))
        if overlap > 0:
            keyword_results.append({
                "chunk": chunk,
                "keyword_score": overlap / len(query_words)
            })

    keyword_results.sort(key=lambda x: x["keyword_score"], reverse=True)

    # Combine and re-rank
    # Using Reciprocal Rank Fusion (RRF)
    combined_scores = {}

    for i, result in enumerate(vector_results):
        chunk_id = result["metadata"]["chunk_id"]
        combined_scores[chunk_id] = combined_scores.get(chunk_id, 0) + 1 / (i + 1)

    for i, result in enumerate(keyword_results[:top_k]):
        chunk_id = result["chunk"]["metadata"]["chunk_id"]
        combined_scores[chunk_id] = combined_scores.get(chunk_id, 0) + 1 / (i + 1)

    # Sort by combined score
    ranked_chunks = sorted(
        [(chunk_id, score) for chunk_id, score in combined_scores.items()],
        key=lambda x: x[1],
        reverse=True
    )

    return [embedded_chunks[chunk_id] for chunk_id, _ in ranked_chunks[:top_k]]

# Use hybrid search
results = hybrid_search("connectivity troubleshooting", endpoint, None, top_k=5)

def rag_with_query_expansion(question, endpoint):
    """Improve retrieval with query expansion"""

    # Generate alternative query formulations
    model = GenerativeModel("gemini-1.0-pro")

    expansion_prompt = f"""Generate 3 alternative ways to ask this question:

Original: {question}

Alternatives (one per line):"""

    response = model.generate_content(expansion_prompt, generation_config={"temperature": 0.7})
    alternative_queries = [q.strip() for q in response.text.split("\n") if q.strip()]

    # Search with all queries
    all_chunks = []
    seen_chunks = set()

    for query in [question] + alternative_queries:
        chunks = search_similar_chunks(query, endpoint, top_k=3)
        for chunk in chunks:
            chunk_id = chunk["metadata"]["chunk_id"]
            if chunk_id not in seen_chunks:
                all_chunks.append(chunk)
                seen_chunks.add(chunk_id)

    # Build context and generate
    context = "\n\n".join([
        f"[Source {i+1}]\n{chunk['text']}"
        for i, chunk in enumerate(all_chunks[:10])
    ])

    prompt = f"""Based on the context, answer the question.

Context:
{context}

Question: {question}

Answer:"""

    response = model.generate_content(prompt)
    return response.text

# Use query expansion
answer = rag_with_query_expansion(
    "What should I do if the device won't connect?",
    endpoint
)
print(f"Answer: {answer}")

def evaluate_rag_system(test_questions, endpoint):
    """Evaluate RAG system quality"""

    results = []

    for question_data in test_questions:
        question = question_data["question"]
        expected_answer = question_data.get("expected_answer")

        # Get RAG result
        result = rag_query(question, endpoint)

        # Evaluate retrieval
        relevant_retrieved = any(
            source["chunk_id"] in question_data.get("relevant_chunks", [])
            for source in result["sources"]
        )

        # Evaluate answer quality (using model)
        model = GenerativeModel("gemini-1.0-pro")
        eval_prompt = f"""Evaluate if the answer correctly addresses the question.

Question: {question}
Answer: {result['answer']}
Expected: {expected_answer}

Is the answer correct and complete? (YES/NO):"""

        eval_response = model.generate_content(eval_prompt, generation_config={"temperature": 0.1})

        results.append({
            "question": question,
            "retrieval_success": relevant_retrieved,
            "answer_quality": eval_response.text.strip(),
            "num_sources": len(result["sources"])
        })

    # Calculate metrics
    retrieval_success_rate = sum(r["retrieval_success"] for r in results) / len(results)
    print(f"Retrieval Success Rate: {retrieval_success_rate:.2%}")

    return results

# Test RAG system
test_cases = [
    {
        "question": "How do I reset the device?",
        "expected_answer": "Press and hold the power button for 10 seconds",
        "relevant_chunks": [5, 12, 23]
    },
    # ... more test cases
]

# evaluation_results = evaluate_rag_system(test_cases, endpoint)

from google.cloud import aiplatform

# Initialize with VPC configuration
aiplatform.init(
    project="my-project",
    location="us-central1",
)

# Deploy model with private endpoint
def deploy_private_endpoint(model, vpc_network):
    """Deploy model to private VPC endpoint"""

    # Create private endpoint
    endpoint = aiplatform.Endpoint.create(
        display_name="private-genai-endpoint",
        network=f"projects/{PROJECT_NUMBER}/global/networks/{vpc_network}",
        encryption_spec_key_name=f"projects/{PROJECT_ID}/locations/{LOCATION}/keyRings/{KEY_RING}/cryptoKeys/{KEY}"
    )

    # Deploy model
    endpoint.deploy(
        model=model,
        deployed_model_display_name="private-model",
        machine_type="n1-standard-4",
        min_replica_count=2,
        max_replica_count=10,
        traffic_split={"0": 100},
        enable_access_logging=True
    )

    return endpoint

# Use private endpoint
private_endpoint = deploy_private_endpoint(
    model,
    vpc_network="my-vpc"
)

# Prediction from within VPC
prediction = private_endpoint.predict(instances=data)

# Configure VPC peering for Vertex AI
# (Done via gcloud or Console)

# gcloud compute networks peerings create vertex-ai-peering \
#     --network=my-vpc \
#     --peer-project=vertex-ai-project \
#     --peer-network=vertex-ai-network \
#     --import-custom-routes \
#     --export-custom-routes

# Use Private Service Connect
from google.cloud import aiplatform

# Create endpoint with PSC
psc_endpoint = aiplatform.Endpoint.create(
    display_name="psc-genai-endpoint",
    private_service_connect_config={
        "enable_private_service_connect": True,
        "project_allowlist": [PROJECT_ID]
    }
)

# Access via Private Service Connect attachment
# Requests stay within Google's network

# Specify data residency requirements
import vertexai

# Initialize with specific region
vertexai.init(
    project="my-project",
    location="europe-west4"  # Data stays in Europe
)

# Load model from specific region
from vertexai.language_models import TextGenerationModel

model = TextGenerationModel.from_pretrained(
    "text-bison@002"
)

# Predictions processed in specified region
response = model.predict(
    prompt="Sensitive data processing...",
    # Data processed in europe-west4
)

# Store tuned models with residency constraints
tuning_job = model.tune_model(
    training_data="gs://eu-bucket/training-data.jsonl",
    tuning_job_location="europe-west4",
    tuned_model_location="europe-west4",  # Model stays in EU
    model_display_name="eu-compliant-model"
)

from google.cloud import aiplatform
from google.cloud import kms_v1

# Customer-Managed Encryption Keys (CMEK)
def deploy_with_cmek(model):
    """Deploy model with CMEK encryption"""

    # KMS key resource name
    kms_key = f"projects/{PROJECT_ID}/locations/{LOCATION}/keyRings/{KEY_RING}/cryptoKeys/{KEY}"

    # Create endpoint with CMEK
    endpoint = aiplatform.Endpoint.create(
        display_name="cmek-encrypted-endpoint",
        encryption_spec_key_name=kms_key
    )

    # Deploy model
    endpoint.deploy(
        model=model,
        deployed_model_display_name="encrypted-model",
        machine_type="n1-standard-4"
    )

    return endpoint

# VPC Service Controls
# Define security perimeter
from google.cloud import accesscontextmanager_v1

def create_service_perimeter():
    """Create VPC Service Controls perimeter"""

    client = accesscontextmanager_v1.AccessContextManagerClient()

    perimeter = {
        "name": f"accessPolicies/{POLICY_ID}/servicePerimeters/vertex_ai_perimeter",
        "title": "Vertex AI Perimeter",
        "perimeter_type": "PERIMETER_TYPE_REGULAR",
        "status": {
            "resources": [
                f"projects/{PROJECT_NUMBER}"
            ],
            "restricted_services": [
                "aiplatform.googleapis.com"
            ],
            "vpc_accessible_services": {
                "enable_restriction": True,
                "allowed_services": [
                    "aiplatform.googleapis.com",
                    "storage.googleapis.com"
                ]
            }
        }
    }

    # Create perimeter
    # operation = client.create_service_perimeter(...)
    return perimeter

# Audit logging
def enable_audit_logging():
    """Enable comprehensive audit logging"""

    # Cloud Logging configuration
    logging_config = {
        "log_type": "DATA_ACCESS",
        "exempted_members": []
    }

    # All Vertex AI API calls logged
    # View in Cloud Logging:
    # resource.type="aiplatform.googleapis.com/Endpoint"
    # protoPayload.methodName="google.cloud.aiplatform.v1.PredictionService.Predict"

# IAM roles for Vertex AI Gen AI
from google.cloud import resourcemanager_v3

def configure_iam_for_genai():
    """Configure IAM roles for Gen AI workloads"""

    # Predefined roles
    roles = {
        "roles/aiplatform.user": "Use Vertex AI resources",
        "roles/aiplatform.admin": "Full Vertex AI administration",
        "roles/aiplatform.modelUser": "Use deployed models",
        "roles/aiplatform.viewer": "View Vertex AI resources",
        "roles/aiplatform.featurestoreUser": "Access feature store",
        "roles/aiplatform.customCodeServiceAgent": "Custom code execution"
    }

    # Grant model user access to service account
    # gcloud projects add-iam-policy-binding PROJECT_ID \
    #     --member="serviceAccount:genai-app@PROJECT_ID.iam.gserviceaccount.com" \
    #     --role="roles/aiplatform.user"

    # Fine-grained permissions with custom role
    custom_permissions = [
        "aiplatform.endpoints.predict",
        "aiplatform.endpoints.get",
        "aiplatform.models.get"
    ]

    # Create custom role
    # gcloud iam roles create genaiModelUser \
    #     --project=PROJECT_ID \
    #     --permissions=aiplatform.endpoints.predict,aiplatform.endpoints.get

# Workload Identity for GKE
def configure_workload_identity():
    """Configure Workload Identity for Vertex AI access from GKE"""

    # Link Kubernetes SA to Google SA
    # gcloud iam service-accounts add-iam-policy-binding \
    #     genai-app@PROJECT_ID.iam.gserviceaccount.com \
    #     --role roles/iam.workloadIdentityUser \
    #     --member "serviceAccount:PROJECT_ID.svc.id.goog[NAMESPACE/KSA_NAME]"

    # Use from pod
    k8s_manifest = """
apiVersion: v1
kind: ServiceAccount
metadata:
  name: genai-app-sa
  annotations:
    iam.gke.io/gcp-service-account: genai-app@PROJECT_ID.iam.gserviceaccount.com
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: genai-app
spec:
  template:
    spec:
      serviceAccountName: genai-app-sa
      containers:
      - name: app
        image: gcr.io/PROJECT_ID/genai-app:latest
"""

# Implement rate limiting for Gen AI endpoints
from functools import wraps
import time
from collections import deque

class RateLimiter:
    """Token bucket rate limiter"""

    def __init__(self, requests_per_minute=60):
        self.requests_per_minute = requests_per_minute
        self.requests = deque()

    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()

            # Remove requests older than 1 minute
            while self.requests and self.requests[0] < now - 60:
                self.requests.popleft()

            # Check if rate limit exceeded
            if len(self.requests) >= self.requests_per_minute:
                sleep_time = 60 - (now - self.requests[0])
                if sleep_time > 0:
                    time.sleep(sleep_time)

            # Record request
            self.requests.append(time.time())

            return func(*args, **kwargs)

        return wrapper

# Use rate limiter
@RateLimiter(requests_per_minute=60)
def call_genai_model(prompt):
    model = TextGenerationModel.from_pretrained("text-bison@002")
    return model.predict(prompt)

# Monitor quotas
from google.cloud import monitoring_v3

def check_quota_usage():
    """Monitor Vertex AI API quota usage"""

    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{PROJECT_ID}"

    # Query quota metrics
    interval = monitoring_v3.TimeInterval(
        {
            "end_time": {"seconds": int(time.time())},
            "start_time": {"seconds": int(time.time()) - 3600},
        }
    )

    results = client.list_time_series(
        request={
            "name": project_name,
            "filter": 'metric.type="serviceruntime.googleapis.com/quota/rate/net_usage" AND resource.service="aiplatform.googleapis.com"',
            "interval": interval,
            "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
        }
    )

    for result in results:
        print(f"Quota: {result.metric.labels['quota_metric']}")
        for point in result.points:
            print(f"  Usage: {point.value.int64_value}")

# check_quota_usage()

"""
Scenario: Build a Q&A system for internal company documentation
Requirements:
- Process 10,000+ internal documents
- Provide accurate answers with citations
- Support multiple document formats (PDF, DOCX, HTML)
- Maintain data privacy and security
- Scale to handle 1000+ concurrent users
"""

from vertexai.language_models import TextEmbeddingModel
from vertexai.generative_models import GenerativeModel
from google.cloud import storage, aiplatform
import PyPDF2
from docx import Document

class EnterpriseDocumentQA:
    def __init__(self, project_id, location="us-central1"):
        self.project_id = project_id
        self.location = location

        # Initialize models
        aiplatform.init(project=project_id, location=location)
        self.embedding_model = TextEmbeddingModel.from_pretrained("textembedding-gecko@003")
        self.generation_model = GenerativeModel("gemini-1.0-pro")

        # Storage
        self.storage_client = storage.Client()
        self.document_chunks = []

    def process_documents(self, gcs_folder_uri):
        """Process all documents in GCS folder"""

        # Extract bucket and prefix
        bucket_name = gcs_folder_uri.replace("gs://", "").split("/")[0]
        prefix = "/".join(gcs_folder_uri.replace("gs://", "").split("/")[1:])

        bucket = self.storage_client.bucket(bucket_name)
        blobs = bucket.list_blobs(prefix=prefix)

        for blob in blobs:
            if blob.name.endswith('.pdf'):
                text = self._extract_pdf(blob)
            elif blob.name.endswith('.docx'):
                text = self._extract_docx(blob)
            elif blob.name.endswith('.html'):
                text = self._extract_html(blob)
            else:
                continue

            # Chunk and embed
            chunks = self._chunk_document(text, blob.name)
            self.document_chunks.extend(chunks)

        # Create vector index
        self._create_vector_index()

    def _chunk_document(self, text, source):
        """Chunk document intelligently"""
        # Split by paragraphs
        paragraphs = text.split('\n\n')

        chunks = []
        current_chunk = ""

        for para in paragraphs:
            if len(current_chunk) + len(para) < 1000:
                current_chunk += para + "\n\n"
            else:
                if current_chunk:
                    chunks.append({
                        "text": current_chunk.strip(),
                        "source": source
                    })
                current_chunk = para + "\n\n"

        if current_chunk:
            chunks.append({"text": current_chunk.strip(), "source": source})

        return chunks

    def answer_question(self, question):
        """Answer question using RAG"""

        # Retrieve relevant chunks
        relevant_chunks = self._retrieve_chunks(question, top_k=5)

        # Build context
        context = "\n\n".join([
            f"[Document: {chunk['source']}]\n{chunk['text']}"
            for chunk in relevant_chunks
        ])

        # Generate answer
        prompt = f"""You are an AI assistant helping employees find information in company documents.
Answer the question based on the provided context. Include the document source in your answer.
If the information is not in the context, say so.

Context:
{context}

Question: {question}

Answer:"""

        response = self.generation_model.generate_content(
            prompt,
            generation_config={
                "temperature": 0.2,
                "max_output_tokens": 1024
            }
        )

        return {
            "answer": response.text,
            "sources": [chunk["source"] for chunk in relevant_chunks]
        }

# Usage
qa_system = EnterpriseDocumentQA("my-project")
qa_system.process_documents("gs://my-bucket/internal-docs/")

result = qa_system.answer_question("What is our remote work policy?")
print(f"Answer: {result['answer']}")
print(f"Sources: {result['sources']}")

"""
Scenario: Build an AI assistant for code generation and review
Requirements:
- Generate code from natural language descriptions
- Review code for bugs and improvements
- Explain complex code sections
- Suggest optimizations
- Support multiple programming languages
"""

from vertexai.language_models import CodeGenerationModel, CodeChatModel
from vertexai.generative_models import GenerativeModel

class CodeAssistant:
    def __init__(self):
        self.code_model = CodeGenerationModel.from_pretrained("code-bison@002")
        self.code_chat = CodeChatModel.from_pretrained("codechat-bison@002")
        self.review_model = GenerativeModel("gemini-1.0-pro")

    def generate_code(self, description, language="python"):
        """Generate code from description"""

        prompt = f"""# Language: {language}
# Task: {description}
# Generate complete, production-ready code with error handling and documentation

"""

        response = self.code_model.predict(
            prefix=prompt,
            temperature=0.2,
            max_output_tokens=2048
        )

        return response.text

    def review_code(self, code, focus_areas=None):
        """Review code for issues and improvements"""

        focus = focus_areas if focus_areas else ["bugs", "performance", "security", "best practices"]

        prompt = f"""Review this code and provide feedback on: {', '.join(focus)}

Code:

"""
Scenario: Build a multi-lingual customer support chatbot
Requirements:
- Support 20+ languages
- Access company knowledge base
- Escalate complex issues to humans
- Track conversation history
- Provide sentiment analysis
"""

from vertexai.language_models import ChatModel
from vertexai.generative_models import GenerativeModel
from google.cloud import translate_v2 as translate
from google.cloud import firestore

class MultiLingualSupportBot:
    def __init__(self, project_id):
        self.project_id = project_id
        self.chat_model = ChatModel.from_pretrained("chat-bison@002")
        self.gemini_model = GenerativeModel("gemini-1.0-pro")
        self.translate_client = translate.Client()
        self.db = firestore.Client()

    def start_conversation(self, user_id, language="en"):
        """Start new conversation"""

        conversation = {
            "user_id": user_id,
            "language": language,
            "messages": [],
            "sentiment_scores": [],
            "escalated": False
        }

        doc_ref = self.db.collection("conversations").document()
        doc_ref.set(conversation)

        return doc_ref.id

    def send_message(self, conversation_id, message):
        """Process user message"""

        # Get conversation
        doc_ref = self.db.collection("conversations").document(conversation_id)
        conversation = doc_ref.get().to_dict()

        # Detect language if needed
        if conversation["language"] == "auto":
            detection = self.translate_client.detect_language(message)
            conversation["language"] = detection["language"]

        # Translate to English for processing
        if conversation["language"] != "en":
            translated = self.translate_client.translate(
                message,
                target_language="en"
            )
            message_en = translated["translatedText"]
        else:
            message_en = message

        # Analyze sentiment
        sentiment_score = self._analyze_sentiment(message_en)
        conversation["sentiment_scores"].append(sentiment_score)

        # Check if escalation needed
        if sentiment_score < -0.5 or "speak to manager" in message_en.lower():
            conversation["escalated"] = True
            response = "I understand you're frustrated. Let me connect you with a human agent."

        else:
            # Generate response
            response = self._generate_response(message_en, conversation["messages"])

        # Translate response back
        if conversation["language"] != "en":
            response = self.translate_client.translate(
                response,
                target_language=conversation["language"]
            )["translatedText"]

        # Update conversation
        conversation["messages"].append({
            "user": message,
            "bot": response,
            "timestamp": firestore.SERVER_TIMESTAMP
        })

        doc_ref.set(conversation)

        return {
            "response": response,
            "escalated": conversation["escalated"],
            "sentiment": sentiment_score
        }

    def _generate_response(self, message, history):
        """Generate contextual response"""

        # Build conversation context
        context = "You are a helpful customer support agent. Be empathetic and professional."
        if history:
            context += "\n\nPrevious conversation:\n"
            for msg in history[-3:]:  # Last 3 exchanges
                context += f"User: {msg['user']}\nAgent: {msg['bot']}\n"

        chat = self.chat_model.start_chat(context=context)
        response = chat.send_message(message)

        return response.text

    def _analyze_sentiment(self, text):
        """Analyze message sentiment"""

        prompt = f"""Analyze the sentiment of this message.
Return a score from -1 (very negative) to +1 (very positive).

Message: {text}

Score (number only):"""

        response = self.gemini_model.generate_content(
            prompt,
            generation_config={"temperature": 0.1, "max_output_tokens": 10}
        )

        try:
            return float(response.text.strip())
        except:
            return 0.0

# Usage
bot = MultiLingualSupportBot("my-project")

# Start conversation
conv_id = bot.start_conversation("user123", language="es")

# Send messages
response1 = bot.send_message(conv_id, "Hola, tengo un problema con mi pedido")
print(f"Bot: {response1['response']}")
print(f"Sentiment: {response1['sentiment']}")

response2 = bot.send_message(conv_id, "El producto no funciona correctamente")
print(f"Bot: {response2['response']}")

"""
Scenario: Build an AI-powered content moderation system
Requirements:
- Detect harmful content (hate speech, violence, etc.)
- Multi-modal (text and images)
- Provide confidence scores
- Support custom policies
- Handle high throughput (10K+ requests/sec)
"""

from vertexai.generative_models import GenerativeModel
from vertexai.vision_models import ImageTextModel
import vertexai.preview.generative_models as generative_models

class ContentModerationSystem:
    def __init__(self):
        self.text_model = GenerativeModel("gemini-1.0-pro")
        self.vision_model = GenerativeModel("gemini-1.0-pro-vision")

    def moderate_text(self, text, custom_policies=None):
        """Moderate text content"""

        # Use built-in safety settings
        safety_settings = {
            generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
            generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
            generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
            generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
        }

        response = self.text_model.generate_content(
            text,
            safety_settings=safety_settings
        )

        # Extract safety ratings
        safety_ratings = {}
        for rating in response.candidates[0].safety_ratings:
            safety_ratings[rating.category.name] = {
                "probability": rating.probability.name,
                "blocked": rating.blocked
            }

        # Custom policy check
        policy_violations = []
        if custom_policies:
            policy_violations = self._check_custom_policies(text, custom_policies)

        return {
            "approved": not response.candidates[0].finish_reason.name.startswith("SAFETY"),
            "safety_ratings": safety_ratings,
            "policy_violations": policy_violations,
            "confidence": self._calculate_confidence(safety_ratings)
        }

    def moderate_image(self, image_uri, text_overlay=None):
        """Moderate image content"""

        prompt = """Analyze this image for inappropriate content.
Check for:
1. Nudity or sexual content
2. Violence or gore
3. Hate symbols
4. Illegal activities

Provide a safety rating (SAFE/CAUTION/UNSAFE) and explanation:"""

        parts = [prompt, Part.from_uri(image_uri, mime_type="image/jpeg")]
        if text_overlay:
            parts.append(f"\nText overlay: {text_overlay}")

        response = self.vision_model.generate_content(parts)

        # Parse response
        result_text = response.text.lower()
        if "unsafe" in result_text:
            rating = "UNSAFE"
        elif "caution" in result_text:
            rating = "CAUTION"
        else:
            rating = "SAFE"

        return {
            "approved": rating == "SAFE",
            "rating": rating,
            "explanation": response.text,
            "safety_ratings": {
                rating.category.name: rating.probability.name
                for rating in response.candidates[0].safety_ratings
            }
        }

    def _check_custom_policies(self, text, policies):
        """Check against custom company policies"""

        violations = []

        for policy in policies:
            prompt = f"""Check if this text violates the following policy:

Policy: {policy['description']}
Examples of violations: {policy.get('examples', [])}

Text: {text}

Does this violate the policy? (YES/NO):
Explanation:"""

            response = self.text_model.generate_content(
                prompt,
                generation_config={"temperature": 0.1}
            )

            if response.text.strip().upper().startswith("YES"):
                violations.append({
                    "policy": policy["name"],
                    "explanation": response.text
                })

        return violations

    def _calculate_confidence(self, safety_ratings):
        """Calculate overall confidence score"""

        # Simplified confidence calculation
        high_risk_count = sum(
            1 for rating in safety_ratings.values()
            if rating["probability"] in ["HIGH", "MEDIUM"]
        )

        if high_risk_count == 0:
            return 0.95
        elif high_risk_count == 1:
            return 0.75
        else:
            return 0.50

# Usage
moderator = ContentModerationSystem()

# Moderate text
result = moderator.moderate_text(
    "This is a sample message to moderate",
    custom_policies=[
        {
            "name": "no_competitor_mentions",
            "description": "Do not mention competitor products",
            "examples": ["Check out AcmeCorp instead"]
        }
    ]
)

print(f"Approved: {result['approved']}")
print(f"Confidence: {result['confidence']}")
print(f"Safety ratings: {result['safety_ratings']}")