The 2026 AI Landscape: A Hacker's Deep Dive 🤖

import torch
import torch.nn.functional as F
import math

def scaled_dot_product_attention(Q, K, V, mask=None):
    """
    Core self-attention mechanism.
    Q, K, V: (batch_size, seq_len, d_k)
    """
    d_k = Q.size(-1)
    
    # Compute attention scores
    scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k)
    
    # Apply optional mask (for decoder / causal attention)
    if mask is not None:
        scores = scores.masked_fill(mask == 0, float('-inf'))
    
    # Softmax to get attention weights
    attn_weights = F.softmax(scores, dim=-1)
    
    # Weighted sum of values
    output = torch.matmul(attn_weights, V)
    return output, attn_weights

# Example: 2 sentences, 4 tokens, 8-dim embeddings
batch_size, seq_len, d_model = 2, 4, 8
d_k = 8

Q = torch.randn(batch_size, seq_len, d_k)
K = torch.randn(batch_size, seq_len, d_k)
V = torch.randn(batch_size, seq_len, d_k)

output, weights = scaled_dot_product_attention(Q, K, V)
print(f"Output shape: {output.shape}")        # (2, 4, 8)
print(f"Attention weights: {weights.shape}")  # (2, 4, 4)

import torch.nn as nn

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()
        assert d_model % num_heads == 0
        self.d_k = d_model // num_heads
        self.num_heads = num_heads
        
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)
    
    def split_heads(self, x):
        B, T, d = x.shape
        # (B, T, d) -> (B, num_heads, T, d_k)
        return x.view(B, T, self.num_heads, self.d_k).transpose(1, 2)
    
    def forward(self, x, mask=None):
        Q = self.split_heads(self.W_q(x))
        K = self.split_heads(self.W_k(x))
        V = self.split_heads(self.W_v(x))
        
        attn_out, _ = scaled_dot_product_attention(Q, K, V, mask)
        
        # Merge heads: (B, num_heads, T, d_k) -> (B, T, d_model)
        B, H, T, d_k = attn_out.shape
        attn_out = attn_out.transpose(1, 2).contiguous().view(B, T, H * d_k)
        
        return self.W_o(attn_out)

Thought: I need to find the current population of Karachi.
Action: search("Karachi population 2026")
Observation: Karachi population is approximately 16.5 million.
Thought: Now I can answer the question.
Answer: Karachi has approximately 16.5 million people.

from openai import OpenAI
import json

client = OpenAI()

# Define tools the agent can use
tools = [
    {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "Search the web for current information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function", 
        "function": {
            "name": "run_python",
            "description": "Execute Python code and return the result",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "Python code to run"}
                },
                "required": ["code"]
            }
        }
    }
]

def run_react_agent(task: str, max_steps: int = 5):
    messages = [{"role": "user", "content": task}]
    
    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        msg = response.choices[0].message
        
        # No tool call = final answer
        if not msg.tool_calls:
            return msg.content
        
        # Execute tool calls
        messages.append(msg)
        for tool_call in msg.tool_calls:
            result = execute_tool(tool_call.function.name, 
                                  json.loads(tool_call.function.arguments))
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result)
            })
    
    return "Max steps reached"

                    ┌─────────────────┐
                    │   Orchestrator   │
                    │   (Planner LLM) │
                    └────────┬────────┘
            ┌───────────────┼───────────────┐
            ▼               ▼               ▼
    ┌──────────────┐ ┌─────────────┐ ┌──────────────┐
    │  Researcher  │ │   Coder     │ │   Reviewer   │
    │   Agent      │ │   Agent     │ │   Agent      │
    └──────────────┘ └─────────────┘ └──────────────┘

User Query
    │
    ▼
Embed Query (vector)
    │
    ▼
Similarity Search → Vector DB → Top-K Documents
    │
    ▼
Inject Context into LLM Prompt
    │
    ▼
LLM generates grounded response

from openai import OpenAI
import numpy as np

client = OpenAI()

# Step 1: Embed documents at ingestion time
def embed_text(text: str) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# Step 2: Simple cosine similarity search
def cosine_similarity(a: list, b: list) -> float:
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def retrieve(query: str, documents: list[dict], top_k: int = 3):
    query_vec = embed_text(query)
    scored = [
        (doc, cosine_similarity(query_vec, doc["embedding"]))
        for doc in documents
    ]
    return sorted(scored, key=lambda x: x[1], reverse=True)[:top_k]

# Step 3: Generate grounded answer
def rag_answer(query: str, docs: list[str]) -> str:
    context = "\n\n".join(docs)
    prompt = f"""Answer the question using ONLY the context below.
If the answer isn't in the context, say "I don't know."

Context:
{context}

Question: {query}"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

from typing import Optional

class AgenticRAGPipeline:
    """
    Agentic RAG: Agent decides WHEN and WHAT to retrieve,
    can iterate multiple retrieval rounds, and validates output.
    """
    
    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.client = llm_client
        self.retrieved_docs = []
    
    def should_retrieve_more(self, current_answer: str, query: str) -> bool:
        """Ask the LLM if it needs more context."""
        check_prompt = f"""
Query: {query}
Current draft answer: {current_answer}

Is this answer complete and well-supported? 
Reply with JSON: {{"complete": true/false, "missing": "what's missing"}}
"""
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": check_prompt}],
            response_format={"type": "json_object"}
        )
        import json
        result = json.loads(response.choices[0].message.content)
        return not result["complete"], result.get("missing", "")
    
    def run(self, query: str, max_iterations: int = 3) -> str:
        answer = ""
        for i in range(max_iterations):
            # Retrieve relevant docs
            search_query = query if i == 0 else f"{query} - focusing on: {answer}"
            new_docs = self.retriever.search(search_query, top_k=5)
            self.retrieved_docs.extend(new_docs)
            
            # Generate answer with all accumulated context
            context = "\n---\n".join(self.retrieved_docs)
            answer = self._generate(query, context)
            
            # Check if we need more info
            needs_more, missing = self.should_retrieve_more(answer, query)
            if not needs_more:
                break
                
        return answer
    
    def _generate(self, query: str, context: str) -> str:
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": f"Context:\n{context}\n\nAnswer: {query}"
            }]
        )
        return response.choices[0].message.content

┌─────────────────────────────────────────────────┐
│                  MCP Architecture                │
│                                                  │
│  ┌──────────────┐      MCP Protocol              │
│  │  AI Client   │◄──────────────────►┐           │
│  │ (Claude/GPT) │                    │           │
│  └──────────────┘         ┌──────────┴────────┐  │
│                            │    MCP Server     │  │
│                            │  (your tools)     │  │
│                            │                   │  │
│                            │  ┌─────────────┐  │  │
│                            │  │  Tools      │  │  │
│                            │  │  Resources  │  │  │
│                            │  │  Prompts    │  │  │
│                            │  └─────────────┘  │  │
│                            └───────────────────┘  │
└─────────────────────────────────────────────────┘

# Minimal MCP server using the official Python SDK
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import mcp.types as types

app = Server("my-mcp-server")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="Get the current weather for a city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="run_sql",
            description="Execute a read-only SQL query against our DB",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_weather":
        city = arguments["city"]
        # Call your actual weather API here
        return [TextContent(type="text", text=f"Weather in {city}: 28°C, sunny")]
    
    elif name == "run_sql":
        query = arguments["query"]
        # Execute query safely
        results = execute_readonly_query(query)
        return [TextContent(type="text", text=str(results))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

MCP Layer (Vertical):          A2A Layer (Horizontal):
Agent ◄──tools──► APIs         Agent ◄──collaborate──► Agent
Agent ◄──tools──► Databases    Agent ◄──delegate───► Agent
Agent ◄──tools──► Files        Agent ◄──coordinate──► Agent

{
  "name": "payment-processor-agent",
  "version": "1.2.0",
  "description": "Handles payment processing for e-commerce workflows",
  "endpoint": "https://payments.example.com/a2a",
  "skills": [
    {
      "id": "process_payment",
      "name": "Process Payment",
      "description": "Charge a customer for a transaction",
      "inputModes": ["text", "json"],
      "outputModes": ["json"]
    },
    {
      "id": "refund",
      "name": "Issue Refund",
      "description": "Refund a previously processed payment"
    }
  ],
  "authentication": {
    "schemes": ["Bearer"]
  }
}

┌──────────────────────────────────────────────────┐
│              Enterprise AI Architecture           │
│                                                   │
│    ┌─────────┐   A2A   ┌─────────┐               │
│    │ Agent A │◄───────►│ Agent B │               │
│    └────┬────┘         └────┬────┘               │
│         │ MCP               │ MCP                 │
│    ┌────┴────┐         ┌────┴────┐               │
│    │ Tools & │         │ Tools & │               │
│    │  Data   │         │  Data   │               │
│    └─────────┘         └─────────┘               │
│                                                   │
│  MCP = vertical (agent ↔ tools)                  │
│  A2A = horizontal (agent ↔ agent)                │
└──────────────────────────────────────────────────┘

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    query: str
    documents: list[str]
    answer: str
    needs_more_info: bool

def retrieve_node(state: AgentState) -> AgentState:
    """Retrieve relevant documents."""
    docs = vector_store.search(state["query"], k=5)
    return {"documents": docs}

def generate_node(state: AgentState) -> AgentState:
    """Generate answer from retrieved docs."""
    context = "\n".join(state["documents"])
    answer = llm.invoke(f"Context: {context}\nQuestion: {state['query']}")
    return {"answer": answer, "needs_more_info": False}

def check_node(state: AgentState) -> str:
    """Route: done or need more retrieval?"""
    return "done" if not state["needs_more_info"] else "retrieve"

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_conditional_edges("generate", check_node, {
    "done": END,
    "retrieve": "retrieve"  # Loop back if needed
})

app = workflow.compile()

# Run it
result = app.invoke({"query": "What is the capital of Punjab?", "messages": []})
print(result["answer"])

from crewai import Agent, Task, Crew, Process

# Define specialized agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find accurate, up-to-date information on the given topic",
    backstory="Expert researcher with access to web search and databases",
    verbose=True,
    allow_delegation=False,
    tools=[web_search_tool, wikipedia_tool]
)

writer = Agent(
    role="Technical Writer",
    goal="Write clear, engaging blog posts from research findings",
    backstory="Experienced tech blogger who makes complex topics accessible",
    verbose=True,
    allow_delegation=False
)

# Define tasks
research_task = Task(
    description="Research the latest developments in {topic}. "
                "Find key facts, statistics, and expert opinions.",
    expected_output="A structured research brief with citations",
    agent=researcher
)

writing_task = Task(
    description="Write a 1000-word blog post based on the research brief. "
                "Make it engaging for a technical audience.",
    expected_output="A complete, publication-ready blog post in Markdown",
    agent=writer,
    context=[research_task]  # Uses output from research_task
)

# Assemble and run the crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=2
)

result = crew.kickoff(inputs={"topic": "MCP protocol for AI agents"})
print(result)

# ─── Zero-Shot ──────────────────────────────────────────────────────────────
# No examples. Rely on the model's training.
zero_shot = "Classify the sentiment of this review: 'The app crashes constantly.'"

# ─── Few-Shot ────────────────────────────────────────────────────────────────
# Provide examples to guide the model
few_shot = """
Classify sentiment. Examples:
Input: "Love this product!" → Positive
Input: "Terrible experience." → Negative
Input: "It's okay, nothing special." → Neutral

Now classify: "The battery life is surprisingly good."
"""

# ─── Chain-of-Thought (CoT) ──────────────────────────────────────────────────
# Force step-by-step reasoning for complex tasks
cot = """
Solve this step by step:
A store sells apples for Rs. 50 each. If Ali buys 12 apples with Rs. 700, 
how much change does he get?

Think through it step by step before giving the final answer.
"""

# ─── ReAct Pattern ───────────────────────────────────────────────────────────
# Alternate reasoning and action
react_system = """You are an agent. For each task:
1. Thought: reason about what to do next
2. Action: choose a tool [search | calculate | respond]
3. Observation: note what the tool returned
4. Repeat until you have the final answer.
"""

# ─── Self-Consistency ────────────────────────────────────────────────────────
# Sample multiple reasoning paths, pick the majority answer
import re
from collections import Counter

def self_consistent_answer(question: str, client, n_samples: int = 5) -> str:
    answers = []
    for _ in range(n_samples):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user", 
                "content": f"{question}\nThink step by step."
            }],
            temperature=0.7  # Some variability for diversity
        )
        # Extract final answer (last number or key phrase)
        text = response.choices[0].message.content
        answers.append(text.strip().split("\n")[-1])
    
    # Return most common answer
    return Counter(answers).most_common(1)[0][0]

SYSTEM_PROMPT_TEMPLATE = """
## Role
You are {role_name}, a {expertise_level} specialist in {domain}.

## Objective  
{primary_objective}

## Constraints
- Always cite sources when making factual claims
- If uncertain, say "I'm not sure" rather than guessing  
- Keep responses under {max_length} words unless asked for detail
- Output format: {output_format}

## Context
Today's date: {date}
User's technical level: {user_level}

## Examples
{few_shot_examples}
"""

from openai import OpenAI
import numpy as np

client = OpenAI()

def get_embedding(text: str) -> list[float]:
    return client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    ).data[0].embedding

# Semantic similarity demo
sentences = [
    "Karachi is the largest city in Pakistan",
    "The metropolitan area of Karachi has 16 million people",
    "I like to eat biryani",
    "Python is a programming language"
]

embeddings = [get_embedding(s) for s in sentences]

def cosine_sim(a, b):
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

query_emb = get_embedding("What is the population of Karachi?")

for s, emb in zip(sentences, embeddings):
    score = cosine_sim(query_emb, emb)
    print(f"{score:.3f} | {s}")

# Output:
# 0.812 | Karachi is the largest city in Pakistan  ← high
# 0.798 | The metropolitan area of Karachi...      ← high  
# 0.312 | I like to eat biryani                    ← low
# 0.289 | Python is a programming language         ← low

# Quick start: Chroma (local, perfect for prototyping)
import chromadb
from chromadb.utils import embedding_functions

client = chromadb.Client()
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="YOUR_KEY",
    model_name="text-embedding-3-small"
)

collection = client.create_collection(
    name="knowledge_base",
    embedding_function=openai_ef
)

# Add documents
collection.add(
    documents=[
        "MCP is an open standard by Anthropic for AI tool integration",
        "RAG stands for Retrieval-Augmented Generation",
        "LangGraph is a framework for building stateful agent workflows"
    ],
    ids=["doc1", "doc2", "doc3"]
)

# Query
results = collection.query(
    query_texts=["How do AI agents connect to external tools?"],
    n_results=2
)
print(results["documents"])
# [['MCP is an open standard by Anthropic...', 
#   'LangGraph is a framework...']]