llmintegration（llmintegration 搜索）

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

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

# Basic completion

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

# With system prompt

message = client.messages.create(model="claude-sonnet-4-20250514", 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-4-20250514", 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-4-20250514", max_tokens=1024, tools=tools, messages=[{"role": "user", "content": "What's the weather in San Francisco?"}]) ```

## OpenAI GPT Integration

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

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

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

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-4-20250514"): 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 providers[provider](https://github.com/lobbi-docs/claude/blob/HEAD/.claude/skills/llm-integration/**kwargs) ```

## Prompt Engineering Best Practices

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

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

# Enable extended thinking (Claude Sonnet 4 and Opus 4)

response = client.messages.create(model="claude-sonnet-4-20250514", max_tokens=16000, thinking={"type": "enabled", "budget_tokens": 10000 # Reserve tokens for thinking}, 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

Stream thinking process in real-time

with client.messages.stream(model="claude-sonnet-4-20250514", max_tokens=16000, thinking={"type": "enabled", "budget_tokens": 10000}, 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

@dataclass class ThinkingConfig: enabled: bool = False budget_tokens: Optional[int] = None 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": "enabled", "budget_tokens": self.config.budget_tokens or 10000 }

response = client.messages.create( model=kwargs.get("model", "claude-sonnet-4-20250514"), 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, budget_tokens=8000))

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

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

# XML tags help Claude parse complex inputs

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

Provide output in this format: High|Medium|Low path/to/file.py 123 Clear description How to fix """ ```

### 3. Provide Examples (Few-Shot Prompting)

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

4. Assign Roles for Context

- 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-4-20250514", 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

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-4-20250514", max_tokens=16000, thinking={"type": "enabled", "budget_tokens": 10000}, messages=[{"role": "user", "content": COT_PROMPT.format(problem=complex_problem)}]) ```

6. Prefill Responses for Format Control

# Force JSON output by prefilling assistant response

messages = [{"role": "user", "content": "Extract entities from: 'Apple Inc. hired John Smith as CEO in 2023.'"}, {"role": "assistant", "content": "{"} # Prefill to force JSON]

response = client.messages.create(model="claude-sonnet-4-20250514", max_tokens=1024, messages=messages)

# Claude will complete the JSON starting with "{"

json_output = "{" + response.content[0].text ```

### 7. Long Context Best Practices

For Claude's 200k token context window

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

def count_tokens_anthropic(text: str, model: str = "claude-sonnet-4-20250514") -> 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-4-20250514") -> 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

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

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

# Model-specific limits self.limits = { "claude-opus-4-20250514": {"context": 200000, "output": 16384}, "claude-sonnet-4-20250514": {"context": 200000, "output": 16384}, "claude-haiku-3-5-20250514": {"context": 200000, "output": 8192}, }

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": 16384})

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-4-20250514")

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

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:

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-4-20250514", "input_tokens": 1500, "output_tokens": 800, "thinking_tokens": 3000, "extended_thinking": True, "task": "security_audit"}) ```

Model Selection Guide

Claude Models

Extended Thinking: Only available on Sonnet 4 and Opus 4

When to Use Each Model

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-4-20250514")

Ensure model supports extended thinking if requested

if use_thinking and "haiku" in model: model = "claude-sonnet-4-20250514"

return model

Performance Optimization Tips

1. Prompt Caching (Reduce Costs)

# Cache large system prompts or context

response = client.messages.create(model="claude-sonnet-4-20250514", 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

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

Always use streaming for long-running tasks

with client.messages.stream(model="claude-sonnet-4-20250514", 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 ```