LLM Integration Skill

Provides comprehensive LLM integration capabilities for the Golden Armada AI Agent Fleet Platform, including advanced features like extended thinking, sophisticated prompt engineering, and intelligent token budget management.

When to Use This Skill

Activate this skill when working with:

• Claude/Anthropic API integration
• OpenAI GPT integration
• Google Gemini integration
• Ollama local models
• Multi-model orchestration
• Prompt engineering

Anthropic Claude Integration

```python import anthropic

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

Basic completion

message = client.messages.create( model="claude-sonnet-5", max_tokens=1024, messages=[ {"role": "user", "content": "Hello, Claude!"} ] ) print(message.content[0].text)

With system prompt

message = client.messages.create( model="claude-sonnet-5", max_tokens=1024, system="You are a helpful coding assistant.", messages=[ {"role": "user", "content": "Write a Python function to sort a list."} ] )

Streaming

with client.messages.stream( model="claude-sonnet-5", max_tokens=1024, messages=[{"role": "user", "content": "Tell me a story."}] ) as stream: for text in stream.text_stream: print(text, end="", flush=True)

Tool use

tools = [ { "name": "get_weather", "description": "Get the current weather in a location", "input_schema": { "type": "object", "properties": { "location": {"type": "string", "description": "The city and state"} }, "required": ["location"] } } ]

message = client.messages.create( model="claude-sonnet-5", max_tokens=1024, tools=tools, messages=[{"role": "user", "content": "What's the weather in San Francisco?"}] ) ```

OpenAI GPT Integration

```python from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

Basic completion

response = client.chat.completions.create( model="gpt-4", messages=[ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hello!"} ] ) print(response.choices[0].message.content)

Streaming

stream = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "Write a poem."}], stream=True ) for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="")

Function calling

functions = [ { "name": "get_weather", "description": "Get the current weather", "parameters": { "type": "object", "properties": { "location": {"type": "string"} }, "required": ["location"] } } ]

response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "Weather in NYC?"}], functions=functions, function_call="auto" ) ```

Google Gemini Integration

```python import google.generativeai as genai

genai.configure(api_key=os.environ["GOOGLE_API_KEY"])

model = genai.GenerativeModel('gemini-pro')

Basic generation

response = model.generate_content("Explain quantum computing") print(response.text)

Chat

chat = model.start_chat(history=[]) response = chat.send_message("Hello!") print(response.text)

Streaming

response = model.generate_content("Tell me a story", stream=True) for chunk in response: print(chunk.text, end="") ```

Ollama Local Models

```python import ollama

Basic completion

response = ollama.chat( model='llama2', messages=[ {'role': 'user', 'content': 'Hello!'} ] ) print(response['message']['content'])

Streaming

stream = ollama.chat( model='llama2', messages=[{'role': 'user', 'content': 'Tell me a story.'}], stream=True ) for chunk in stream: print(chunk['message']['content'], end='')

Pull model

ollama.pull('llama2')

List models

models = ollama.list() ```

Multi-Model Abstraction

```python from abc import ABC, abstractmethod from typing import Generator

class LLMProvider(ABC): @abstractmethod def generate(self, prompt: str, **kwargs) -> str: pass

@abstractmethod
def stream(self, prompt: str, **kwargs) -> Generator[str, None, None]:
    pass

class ClaudeProvider(LLMProvider): def init(self, api_key: str, model: str = "claude-sonnet-5"): self.client = anthropic.Anthropic(api_key=api_key) self.model = model

def generate(self, prompt: str, **kwargs) -> str:
    message = self.client.messages.create(
        model=self.model,
        max_tokens=kwargs.get('max_tokens', 1024),
        messages=[{"role": "user", "content": prompt}]
    )
    return message.content[0].text

def stream(self, prompt: str, **kwargs) -> Generator[str, None, None]:
    with self.client.messages.stream(
        model=self.model,
        max_tokens=kwargs.get('max_tokens', 1024),
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for text in stream.text_stream:
            yield text

class LLMFactory: @staticmethod def create(provider: str, **kwargs) -> LLMProvider: providers = { 'claude': ClaudeProvider, 'gpt': GPTProvider, 'gemini': GeminiProvider, 'ollama': OllamaProvider } return providersprovider ```

Prompt Engineering Best Practices

```python

Structured prompts

SYSTEM_PROMPT = """You are a helpful coding assistant.

Guidelines:

1. Write clean, well-documented code
2. Follow best practices
3. Explain your reasoning """

Few-shot examples

FEW_SHOT_PROMPT = """Convert natural language to SQL.

Example 1: Input: Get all users Output: SELECT * FROM users;

Example 2: Input: Count active orders Output: SELECT COUNT(*) FROM orders WHERE status = 'active';

Input: {user_input} Output:"""

Chain of thought

COT_PROMPT = """Solve this step by step: {problem}

Let's think through this: 1.""" ```

Extended Thinking Integration

Extended thinking enables Claude models to "think" before responding, improving accuracy on complex tasks like coding, math, and scientific reasoning.

When to Use Extended Thinking

• Complex multi-step reasoning tasks
• Code architecture and system design
• Mathematical problem-solving
• Scientific analysis and research
• Strategic planning and decision-making

Cross-reference: See .claude/skills/extended-thinking/SKILL.md for detailed guidance.

Basic Extended Thinking

```python import anthropic

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

Adaptive thinking (current models: Opus 4.6+, Sonnet 5, Fable 5)

Note: budget_tokens is removed on Opus 4.7+/Sonnet 5/Fable 5 (returns 400);

use output_config effort levels to control depth

response = client.messages.create( model="claude-sonnet-5", max_tokens=16000, thinking={"type": "adaptive"}, output_config={"effort": "high"}, messages=[ { "role": "user", "content": "Design a scalable microservices architecture for a multi-tenant SaaS platform" } ] )

Response includes thinking and final text

for block in response.content: if block.type == "thinking": print(f"Thinking: {block.thinking}") elif block.type == "text": print(f"Response: {block.text}") ```

Extended Thinking with Streaming

```python

Stream thinking process in real-time

with client.messages.stream( model="claude-sonnet-5", max_tokens=16000, thinking={"type": "adaptive"}, messages=[ {"role": "user", "content": "Analyze the time complexity of this sorting algorithm..."} ] ) as stream: for event in stream: if event.type == "content_block_start": if event.content_block.type == "thinking": print("\n[Thinking Process]") elif event.content_block.type == "text": print("\n[Final Answer]") elif event.type == "content_block_delta": if event.delta.type == "thinking_delta": print(event.delta.thinking, end="", flush=True) elif event.delta.type == "text_delta": print(event.delta.text, end="", flush=True) ```

Multi-Provider Extended Thinking Abstraction

```python from typing import Optional, Dict, Any from dataclasses import dataclass

@dataclass class ThinkingConfig: enabled: bool = False effort: str = "high" # low | medium | high | xhigh | max show_thinking: bool = True

class ExtendedThinkingProvider: """Abstract extended thinking across providers"""

def __init__(self, provider: str, config: ThinkingConfig):
    self.provider = provider
    self.config = config

def generate_with_thinking(self, prompt: str, **kwargs) -> Dict[str, Any]:
    if self.provider == "claude":
        return self._claude_thinking(prompt, **kwargs)
    elif self.provider == "gpt":
        # Simulate thinking with chain-of-thought
        return self._gpt_cot_thinking(prompt, **kwargs)
    else:
        raise ValueError(f"Provider {self.provider} doesn't support extended thinking")

def _claude_thinking(self, prompt: str, **kwargs) -> Dict[str, Any]:
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    thinking_params = {}
    if self.config.enabled:
        thinking_params["thinking"] = {"type": "adaptive"}
        thinking_params["output_config"] = {"effort": self.config.effort}

    response = client.messages.create(
        model=kwargs.get("model", "claude-sonnet-5"),
        max_tokens=kwargs.get("max_tokens", 16000),
        messages=[{"role": "user", "content": prompt}],
        **thinking_params
    )

    result = {"thinking": None, "response": None}
    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking
        elif block.type == "text":
            result["response"] = block.text

    return result

def _gpt_cot_thinking(self, prompt: str, **kwargs) -> Dict[str, Any]:
    """Use chain-of-thought prompting for GPT models"""
    from openai import OpenAI
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

    cot_prompt = f"""Let's approach this step-by-step:

{prompt}

First, think through the problem systematically, then provide your final answer."""

    response = client.chat.completions.create(
        model=kwargs.get("model", "gpt-4"),
        messages=[{"role": "user", "content": cot_prompt}]
    )

    # Parse thinking from response (heuristic-based)
    content = response.choices[0].message.content
    parts = content.split("\n\n")

    return {
        "thinking": "\n\n".join(parts[:-1]) if len(parts) > 1 else None,
        "response": parts[-1] if parts else content
    }

Usage

thinking_provider = ExtendedThinkingProvider( provider="claude", config=ThinkingConfig(enabled=True, effort="high") )

result = thinking_provider.generate_with_thinking( "Design a distributed caching strategy for a multi-tenant system" )

if result["thinking"]: print(f"Thinking Process:\n{result['thinking']}\n") print(f"Final Answer:\n{result['response']}") ```

Claude Prompt Engineering Best Practices

Based on Anthropic's official guidelines for optimal performance.

1. Clear and Direct Instructions

```python

BAD: Vague request

prompt = "Make this code better"

GOOD: Specific instructions

prompt = """Refactor this Python function to:

1. Use type hints
2. Add comprehensive docstrings
3. Handle edge cases (empty input, None values)
4. Improve variable naming for clarity
5. Add input validation

Code: {code} """ ```

2. Use XML Tags for Structure

```python

XML tags help Claude parse complex inputs

prompt = """Analyze this codebase and identify security vulnerabilities.

Provide output in this format: <vulnerability> <severity>High|Medium|Low</severity> <file>path/to/file.py</file> <line>123</line> <description>Clear description</description> <recommendation>How to fix</recommendation> </vulnerability> """ ```

3. Provide Examples (Few-Shot Prompting)

```python FEW_SHOT_TEMPLATE = """Convert user stories to acceptance criteria.

Now convert this user story: <user_story> {user_input} </user_story> """ ```

4. Assign Roles for Context

```python ROLE_BASED_SYSTEM_PROMPTS = { "code_reviewer": """You are an expert code reviewer with 15 years of experience. Your expertise includes:

• Software architecture and design patterns
• Security best practices (OWASP Top 10)
• Performance optimization
• Code maintainability and readability

When reviewing code:

1. Identify bugs and potential issues
2. Suggest improvements for clarity and performance
3. Check for security vulnerabilities
4. Recommend design pattern improvements
5. Ensure code follows language best practices

Be constructive and specific in your feedback.""",

"architect": """You are a senior software architect specializing in:

• Microservices and distributed systems
• Cloud-native architecture (AWS, GCP, Azure)
• Database design and optimization
• API design (REST, GraphQL, gRPC)
• Security and compliance

When designing systems:

1. Consider scalability and performance

2. Ensure fault tolerance and resilience

3. Design for observability (logging, metrics, tracing)

4. Follow cloud-native best practices

5. Consider cost optimization""",

   "security_expert": """You are a security specialist focused on:

• OWASP Top 10 vulnerabilities
• Authentication and authorization (OAuth, OIDC, JWT)
• Data encryption and privacy
• Secure coding practices
• Compliance (GDPR, HIPAA, SOC2)

When analyzing security:

1. Identify vulnerabilities with severity ratings
2. Provide specific remediation steps
3. Reference security standards and best practices
4. Consider both code-level and architectural security""" }

Usage

message = client.messages.create( model="claude-sonnet-5", max_tokens=4096, system=ROLE_BASED_SYSTEM_PROMPTS["code_reviewer"], messages=[ {"role": "user", "content": f"Review this code:\n\n{code}"} ] ) ```

5. Chain of Thought Prompting

```python COT_PROMPT = """Solve this problem step by step, showing your reasoning at each stage.

Problem: {problem}

Think through this by:

1. Understanding what's being asked
2. Identifying relevant information
3. Breaking down the problem into steps
4. Solving each step
5. Verifying the solution

Show your work for each step."""

For complex reasoning, combine with extended thinking

response = client.messages.create( model="claude-sonnet-5", max_tokens=16000, thinking={"type": "adaptive"}, messages=[ {"role": "user", "content": COT_PROMPT.format(problem=complex_problem)} ] ) ```

6. Structured Outputs for Format Control

```python

NOTE: assistant-turn prefills return a 400 on current models

(Opus 4.6+, Sonnet 5, Fable 5). Use structured outputs instead:

response = client.messages.create( model="claude-sonnet-5", max_tokens=1024, output_config={ "format": { "type": "json_schema", "schema": { "type": "object", "properties": { "entities": {"type": "array", "items": {"type": "string"}} }, "required": ["entities"], "additionalProperties": False } } }, messages=[ {"role": "user", "content": "Extract entities from: 'Apple Inc. hired John Smith as CEO in 2023.'"} ] )

Output is guaranteed to match the schema

json_output = response.content[0].text ```

7. Long Context Best Practices

```python

For Claude's 1M token context window (200k on Haiku 4.5)

LONG_CONTEXT_TEMPLATE = """I'm providing a large codebase for analysis. The most important files for this task are at the END of this message.

<critical_files> {important_files} </critical_files>

Focus primarily on the critical files when answering the task.""" ```

Token Budget Management

Anthropic Token Counting

```python import anthropic

def count_tokens_anthropic(text: str, model: str = "claude-sonnet-5") -> int: """Count tokens using Anthropic's API""" client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# Use count_tokens endpoint
result = client.messages.count_tokens(
    model=model,
    messages=[{"role": "user", "content": text}]
)

return result.input_tokens

def count_tokens_with_system(messages: list, system: str = None, model: str = "claude-sonnet-5") -> dict: """Count tokens including system prompt and messages""" client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

params = {
    "model": model,
    "messages": messages
}
if system:
    params["system"] = system

result = client.messages.count_tokens(**params)

return {
    "input_tokens": result.input_tokens,
    "system_tokens": getattr(result, "system_tokens", 0)
}

```

Smart Token Budget Management

```python from typing import List, Dict import anthropic

class TokenBudgetManager: """Intelligent token budget management for LLM calls"""

def __init__(self, model: str = "claude-sonnet-5"):
    self.model = model
    self.client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    # Model-specific limits
    self.limits = {
        "claude-fable-5": {"context": 1000000, "output": 128000},
        "claude-opus-4-8": {"context": 1000000, "output": 128000},
        "claude-sonnet-5": {"context": 1000000, "output": 128000},
        "claude-haiku-4-5": {"context": 200000, "output": 64000},
    }

def get_budget(self, model: str = None) -> Dict[str, int]:
    """Get token limits for model"""
    model = model or self.model
    return self.limits.get(model, {"context": 200000, "output": 64000})

def check_budget(
    self,
    messages: List[Dict],
    system: str = None,
    max_tokens: int = 4096,
    thinking_tokens: int = 0
) -> Dict[str, any]:
    """Check if request fits within budget"""

    # Count input tokens
    token_count = self.client.messages.count_tokens(
        model=self.model,
        messages=messages,
        system=system
    )

    input_tokens = token_count.input_tokens
    budget = self.get_budget()

    # Calculate total required tokens
    total_required = input_tokens + max_tokens + thinking_tokens

    return {
        "fits_budget": total_required <= budget["context"],
        "input_tokens": input_tokens,
        "requested_output": max_tokens,
        "thinking_budget": thinking_tokens,
        "total_required": total_required,
        "context_limit": budget["context"],
        "remaining": budget["context"] - total_required,
        "utilization_pct": (total_required / budget["context"]) * 100
    }

def optimize_for_budget(
    self,
    messages: List[Dict],
    system: str = None,
    target_output: int = 4096,
    thinking_tokens: int = 0,
    priority_last_n: int = 3
) -> List[Dict]:
    """Truncate messages to fit budget, preserving recent context"""

    budget = self.get_budget()
    available = budget["context"] - target_output - thinking_tokens

    # Always keep system prompt and last N messages
    preserved_messages = messages[-priority_last_n:]

    # Count tokens for preserved content
    preserved_count = self.client.messages.count_tokens(
        model=self.model,
        messages=preserved_messages,
        system=system
    ).input_tokens

    if preserved_count <= available:
        # Try to include earlier messages
        remaining = available - preserved_count
        earlier_messages = messages[:-priority_last_n]

        # Binary search to find how many earlier messages fit
        left, right = 0, len(earlier_messages)
        best_fit = 0

        while left <= right:
            mid = (left + right) // 2
            test_messages = earlier_messages[-mid:] + preserved_messages

            test_count = self.client.messages.count_tokens(
                model=self.model,
                messages=test_messages,
                system=system
            ).input_tokens

            if test_count <= available:
                best_fit = mid
                left = mid + 1
            else:
                right = mid - 1

        return earlier_messages[-best_fit:] + preserved_messages if best_fit > 0 else preserved_messages
    else:
        # Even preserved messages exceed budget, truncate them
        return preserved_messages[-1:]  # Keep at least the last message

def get_recommendations(self, budget_check: Dict) -> List[str]:
    """Get recommendations based on budget utilization"""
    recommendations = []

    util = budget_check["utilization_pct"]

    if util > 90:
        recommendations.append("CRITICAL: Token usage >90%. Consider reducing context or output length.")
        recommendations.append("Enable extended thinking only if necessary for task complexity.")
    elif util > 75:
        recommendations.append("WARNING: Token usage >75%. Monitor context size.")
        recommendations.append("Consider summarizing earlier conversation turns.")
    elif util > 50:
        recommendations.append("Moderate token usage. Budget healthy.")
    else:
        recommendations.append("Low token usage. Budget has plenty of headroom.")

    if budget_check["thinking_budget"] > budget_check["requested_output"]:
        recommendations.append("Thinking budget exceeds output budget. Ensure this is intentional.")

    return recommendations

Usage example

manager = TokenBudgetManager(model="claude-sonnet-5")

messages = [ {"role": "user", "content": "What is Python?"}, {"role": "assistant", "content": "Python is a high-level programming language..."}, {"role": "user", "content": "Write a complex microservices architecture"} ]

system = "You are an expert software architect."

Check budget

budget_check = manager.check_budget( messages=messages, system=system, max_tokens=8192, thinking_tokens=10000 )

print(f"Fits budget: {budget_check['fits_budget']}") print(f"Utilization: {budget_check['utilization_pct']:.1f}%") print(f"Remaining tokens: {budget_check['remaining']}")

Get recommendations

for rec in manager.get_recommendations(budget_check): print(f"- {rec}")

Optimize if needed

if not budget_check["fits_budget"]: optimized_messages = manager.optimize_for_budget( messages=messages, system=system, target_output=8192, thinking_tokens=10000 ) print(f"Reduced from {len(messages)} to {len(optimized_messages)} messages") ```

OpenAI Token Counting

```python import tiktoken

def count_tokens_openai(text: str, model: str = "gpt-4") -> int: """Count tokens for OpenAI models""" encoding = tiktoken.encoding_for_model(model) return len(encoding.encode(text))

def truncate_to_token_limit(text: str, max_tokens: int, model: str = "gpt-4") -> str: """Truncate text to fit token limit""" encoding = tiktoken.encoding_for_model(model) tokens = encoding.encode(text) if len(tokens) <= max_tokens: return text return encoding.decode(tokens[:max_tokens])

def split_by_tokens(text: str, chunk_size: int, model: str = "gpt-4") -> List[str]: """Split text into chunks of specific token size""" encoding = tiktoken.encoding_for_model(model) tokens = encoding.encode(text)

chunks = []
for i in range(0, len(tokens), chunk_size):
    chunk_tokens = tokens[i:i + chunk_size]
    chunks.append(encoding.decode(chunk_tokens))

return chunks

```

Cross-References and Related Skills

Extended Thinking

For complex reasoning tasks requiring deep analysis:

• Skill: .claude/skills/extended-thinking/SKILL.md
• Use when: Multi-step reasoning, architecture design, mathematical proofs

Complex Reasoning

For advanced problem-solving and analysis:

• Skill: .claude/skills/complex-reasoning/SKILL.md
• Use when: System design, optimization problems, strategic planning

Deep Analysis

For comprehensive codebase and system analysis:

• Skill: .claude/skills/deep-analysis/SKILL.md
• Use when: Code review, security audits, performance analysis

Orchestration Integration

When using LLM integration within the Golden Armada orchestration system:

```python from orchestration.agent_activity_logger import AgentActivityLogger

Log LLM API calls for tracking

logger = AgentActivityLogger()

logger.log_activity( agent_name="code-reviewer", activity_type="llm_api_call", details={ "provider": "anthropic", "model": "claude-sonnet-5", "input_tokens": 1500, "output_tokens": 800, "thinking_tokens": 3000, "extended_thinking": True, "task": "security_audit" } ) ```

Model Selection Guide

Claude Models

Thinking: All current models support adaptive thinking ({"type": "adaptive"}); on Fable 5 thinking is always on (omit the thinking param). Manual budget_tokens applies only to older models.

When to Use Each Model

```python MODEL_SELECTION = { "strategic_planning": "claude-opus-4-8", "system_architecture": "claude-opus-4-8", "complex_debugging": "claude-sonnet-5", "code_review": "claude-sonnet-5", "code_generation": "claude-sonnet-5", "documentation": "claude-haiku-4-5", "simple_queries": "claude-haiku-4-5", "batch_processing": "claude-haiku-4-5" }

def select_model(task_type: str, use_thinking: bool = False) -> str: """Select appropriate model based on task""" model = MODEL_SELECTION.get(task_type, "claude-sonnet-5")

# Prefer a stronger model when deep reasoning is requested
if use_thinking and "haiku" in model:
    model = "claude-sonnet-5"

return model

```

Performance Optimization Tips

1. Prompt Caching (Reduce Costs)

```python

Cache large system prompts or context

response = client.messages.create( model="claude-sonnet-5", max_tokens=4096, system=[ { "type": "text", "text": large_system_prompt, "cache_control": {"type": "ephemeral"} # Cache this } ], messages=[ {"role": "user", "content": "Question about the system"} ] ) ```

2. Batch Processing

```python async def process_batch(items: List[str], batch_size: int = 10): """Process items in batches to avoid rate limits""" import asyncio

async def process_item(item: str):
    # Async LLM call
    return await async_llm_call(item)

results = []
for i in range(0, len(items), batch_size):
    batch = items[i:i + batch_size]
    batch_results = await asyncio.gather(*[process_item(item) for item in batch])
    results.extend(batch_results)

    # Rate limiting
    if i + batch_size < len(items):
        await asyncio.sleep(1)

return results

```

3. Streaming for Responsiveness

```python

Always use streaming for long-running tasks

with client.messages.stream( model="claude-sonnet-5", max_tokens=8192, messages=[{"role": "user", "content": complex_task}] ) as stream: for text in stream.text_stream: yield text # Stream to UI or process incrementally ```