0% found this document useful (0 votes)
3 views11 pages

4 Implementation Patterns Code

The document provides a comprehensive guide on implementing AI agents, emphasizing the importance of a well-structured system prompt, clear tool definitions, and robust error handling. It includes patterns for function calling, memory management, planning, structured output, testing, logging, and cost management, along with common mistakes to avoid. A checklist for quick implementation and additional resources for further learning are also provided.

Uploaded by

vishnu prasad j
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
0% found this document useful (0 votes)
3 views11 pages

4 Implementation Patterns Code

The document provides a comprehensive guide on implementing AI agents, emphasizing the importance of a well-structured system prompt, clear tool definitions, and robust error handling. It includes patterns for function calling, memory management, planning, structured output, testing, logging, and cost management, along with common mistakes to avoid. A checklist for quick implementation and additional resources for further learning are also provided.

Uploaded by

vishnu prasad j
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd

# AI Agent Implementation Patterns & Code Guide

## System Prompt Structure (Critical!)

Your system prompt is your most powerful control mechanism. A vague prompt =
unpredictable agent.

### **Good System Prompt Template**

```
You are a [specific role/purpose].

YOUR GOAL:
[Clear, specific objective]

YOUR TOOLS:
- [Tool name]: [What it does, when to use]
- [Tool name]: [What it does, when to use]

YOUR INSTRUCTIONS:
1. [First step]
2. [Second step]
3. [Continue...]

IMPORTANT RULES:
- [Rule 1: When to stop]
- [Rule 2: Error handling]
- [Rule 3: Format requirements]

OUTPUT FORMAT:
[Specify exact format expected]

DO NOT:
- [What not to do]
- [Common mistakes to avoid]
```

### **Example: Research Agent**

```
You are an expert research assistant.

YOUR GOAL:
Find accurate, well-sourced information on any topic and synthesize
it into a clear, structured report.

YOUR TOOLS:
- web_search: Use to find information on the internet
- read_url: Use to read full article content from URLs
- summarize: Use to condense long articles

YOUR INSTRUCTIONS:
1. Search for reliable sources (prioritize: academic, .gov, established news)
2. Read at least 3 credible sources
3. Synthesize information into structured outline
4. Write clear summary with citations

IMPORTANT RULES:
- Stop after finding consensus from 3+ sources
- If sources conflict, note the disagreement
- Only cite sources you actually read
- Include confidence level (high/medium/low)

OUTPUT FORMAT:
## Research Report: [Topic]
- Key Findings:
- Supporting Evidence (with citations):
- Limitations:
- Confidence Level:
```

---

## Tool Definition Patterns

### **Clear Tool Descriptions**

❌ **Bad:**
```python
tools = [
{"name": "search", "description": "search for stuff"}
]
```

✅ **Good:**
```python
tools = [
{
"name": "web_search",
"description": "Search the internet for current information. Use when you
need facts, recent news, or general knowledge.",
"parameters": {
"query": "Search query string (be specific, include key terms)",
"max_results": "Number of results (default: 5, max: 10)"
},
"when_to_use": "When you need current information or facts verification",
"example_queries": [
"stock market closing prices today",
"latest developments in quantum computing 2024"
]
}
]
```

---

## Function Calling Implementation

### **OpenAI Function Calling Pattern**

```python
import openai
from typing import Any

def run_agent(user_message: str):


"""Run agent with function calling loop"""

messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message}
]

tools = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "Search the web",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
}
},
"required": ["query"]
}
}
},
# More tools...
]

# Agent loop
while True:
response = [Link](
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)

# Check if model wants to call a tool


if response.stop_reason == "tool_calls":
tool_calls = response.tool_calls

for tool_call in tool_calls:


tool_name = tool_call.[Link]
tool_args = [Link](tool_call.[Link])

# Execute tool
result = execute_tool(tool_name, tool_args)

# Add to messages
[Link]({
"role": "assistant",
"content": [Link]
})
[Link]({
"role": "user",
"content": f"Tool result: {result}"
})
else:
# Model returned final answer
return [Link]
```
---

## Error Handling Patterns

### **Robust Tool Execution**

```python
def execute_tool(tool_name: str, args: dict, max_retries: int = 3):
"""Execute tool with error handling and retries"""

for attempt in range(max_retries):


try:
if tool_name == "web_search":
return web_search(**args)
elif tool_name == "read_file":
return read_file(**args)
# More tools...

except TimeoutError:
if attempt < max_retries - 1:
[Link](2 ** attempt) # Exponential backoff
continue
return f"Error: Tool timeout after {max_retries} attempts"

except ValueError as e:
return f"Error: Invalid argument - {str(e)}"

except Exception as e:
log_error(f"Tool {tool_name} failed: {str(e)}")
return f"Error: {tool_name} failed - {str(e)}"

return f"Error: {tool_name} failed after {max_retries} retries"


```

---

## Memory Management Patterns

### **Tiered Memory Implementation**

```python
from typing import List, Dict

class Agent:
def __init__(self):
# Working memory (current task context)
self.working_memory = []

# Short-term memory (recent interactions)


self.short_term_memory = []

# Long-term memory vector store


self.long_term_memory = VectorStore()

def add_to_memory(self, message: str, metadata: dict = None):


"""Add message to appropriate memory tier"""

# Always add to working memory


self.working_memory.append(message)
# If working memory gets too large, archive
if len(self.working_memory) > 10:
summary = self.summarize_messages(
self.working_memory[:-5] # Keep last 5
)
self.short_term_memory.append(summary)
self.working_memory = self.working_memory[-5:]

# If important, add to long-term


if metadata and [Link]("important"):
embedding = [Link](message)
self.long_term_memory.add(embedding, message)

def retrieve_relevant_memory(self, query: str, top_k: int = 3):


"""Retrieve relevant past interactions"""

# Get embedding of query


query_embedding = [Link](query)

# Search long-term memory


relevant = self.long_term_memory.search(
query_embedding,
top_k=top_k
)

return relevant

def get_context_for_llm(self):
"""Build context for LLM with all memories"""

context = []

# Recent context
[Link](self.working_memory[-5:])

# Short-term summaries
[Link](self.short_term_memory[-2:])

# Relevant long-term memories


relevant = self.retrieve_relevant_memory(
str(self.working_memory[-1])
)
if relevant:
[Link]("Relevant past interactions:")
[Link](relevant)

return "\n".join(context)
```

---

## Planning Patterns

### **Explicit Planning Before Action**

```python
def plan_solution(goal: str, tools: List[str], context: str):
"""Have LLM create explicit plan before executing"""
planning_prompt = f"""
Goal: {goal}
Available Tools: {', '.join(tools)}
Current Context: {context}

Create a detailed step-by-step plan to achieve this goal.


Format:
Step 1: [What to do] (use tool: [tool_name])
Step 2: [What to do] (use tool: [tool_name])
...

Reasoning: [Why this plan works]


"""

plan = call_llm(planning_prompt)
return parse_plan(plan)

def execute_plan(plan: List[Dict], agent):


"""Execute plan with monitoring"""

results = []
for i, step in enumerate(plan):
print(f"Executing step {i+1}: {step['description']}")

tool_name = step['tool']
tool_args = step['args']

result = agent.execute_tool(tool_name, tool_args)


[Link](result)

# Check if we should continue


if not should_continue(result):
print("Plan interrupted - checking if goal achieved")
if check_goal_achieved([Link], results):
return results
else:
# Replan
plan = plan_solution(
[Link],
[Link],
results
)

return results
```

---

## Structured Output Patterns

### **Using Pydantic for Output Validation**

```python
from pydantic import BaseModel
from typing import List

class ResearchFinding(BaseModel):
title: str
evidence: List[str]
confidence: float # 0.0 to 1.0
source: str

class ResearchReport(BaseModel):
topic: str
findings: List[ResearchFinding]
summary: str
timestamp: str

def get_structured_output(user_query: str) -> ResearchReport:


"""Get structured output from LLM using JSON schema"""

response = [Link](
model="gpt-4o",
messages=[
{"role": "user", "content": user_query}
],
response_format={
"type": "json_schema",
"json_schema": {
"name": "ResearchReport",
"schema": ResearchReport.model_json_schema()
}
}
)

# Parse and validate


report_data = [Link]([Link])
report = ResearchReport(**report_data)

return report
```

---

## Testing & Evaluation Patterns

### **Unit Testing Agent Components**

```python
import pytest

class TestAgentTools:
def test_web_search_returns_valid_format(self):
"""Verify tool returns expected format"""
result = web_search("Python programming")

assert isinstance(result, list)


assert len(result) > 0
assert all('title' in r and 'url' in r for r in result)

def test_tool_handles_errors_gracefully(self):
"""Verify error handling"""
result = web_search("") # Empty query

assert "error" in [Link]() or isinstance(result, str)

class TestAgentBehavior:
def test_agent_planning(self):
"""Test that agent creates reasonable plans"""
agent = Agent(system_prompt=SYSTEM_PROMPT)
plan = agent.create_plan("Find latest AI breakthroughs")

assert len(plan) > 0


assert all('tool' in step for step in plan)

def test_agent_stops_when_goal_met(self):
"""Verify agent doesn't loop indefinitely"""
agent = Agent(max_steps=5)
result = [Link]("What is 2+2?")

assert agent.steps_taken <= 5


assert "4" in result

class TestAgentEvaluation:
def test_llm_as_judge(self):
"""Use another LLM to evaluate output"""
output = [Link]("List top 3 Python frameworks")

judge_prompt = f"""
Is this output accurate and helpful?
Output: {output}

Rate 1-10 and explain.


"""

rating = call_llm(judge_prompt)
assert float([Link]()[0]) >= 7
```

---

## Logging & Observability

### **Comprehensive Logging Setup**

```python
import logging
import json
from datetime import datetime

class AgentLogger:
def __init__(self, agent_name: str):
self.agent_name = agent_name
[Link] = [Link](agent_name)

# File handler
handler = [Link](f"{agent_name}.log")
[Link](
[Link](
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
)
[Link](handler)

def log_llm_call(self, prompt: str, response: str, tokens: int, cost: float):
"""Log LLM API calls"""
[Link]([Link]({
"type": "llm_call",
"prompt_length": len(prompt),
"response_length": len(response),
"tokens_used": tokens,
"cost": cost,
"timestamp": [Link]().isoformat()
}))

def log_tool_call(self, tool_name: str, args: dict, result: str, duration:


float):
"""Log tool executions"""
[Link]([Link]({
"type": "tool_call",
"tool": tool_name,
"args": args,
"result_summary": result[:100],
"duration_ms": duration * 1000,
"timestamp": [Link]().isoformat()
}))

def log_decision(self, context: str, decision: str, reasoning: str):


"""Log agent decisions"""
[Link]([Link]({
"type": "decision",
"context": context,
"decision": decision,
"reasoning": reasoning,
"timestamp": [Link]().isoformat()
}))
```

---

## Cost Management Patterns

### **Token & Cost Tracking**

```python
class CostTracker:
# Model pricing (as of 2024)
PRICING = {
"gpt-4o": {"input": 0.005, "output": 0.015}, # per 1K tokens
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
"claude-3.5-sonnet": {"input": 0.003, "output": 0.015}
}

def __init__(self, model: str, budget: float):


[Link] = model
[Link] = budget
[Link] = 0.0

def calculate_cost(self, input_tokens: int, output_tokens: int):


"""Calculate cost for API call"""
pricing = [Link][[Link]]

input_cost = (input_tokens / 1000) * pricing["input"]


output_cost = (output_tokens / 1000) * pricing["output"]
total = input_cost + output_cost
[Link] += total

return total

def check_budget(self):
"""Check if still within budget"""
remaining = [Link] - [Link]
percent_used = ([Link] / [Link]) * 100

if percent_used > 90:


log_warning(f"Budget {percent_used}% spent: ${[Link]:.4f}")

return remaining > 0

def estimate_agent_cost(self, num_steps: int, avg_tokens: int):


"""Estimate total cost for agent run"""
estimated = (num_steps * avg_tokens / 1000) * [Link][[Link]]
["output"]
return estimated
```

---

## Common Implementation Mistakes

❌ **Mistake:** Vague system prompt


✅ **Fix:** Detailed, specific instructions with examples

❌ **Mistake:** No error handling


✅ **Fix:** Try-except blocks with fallbacks for each tool

❌ **Mistake:** Uncontrolled loops


✅ **Fix:** Max step limit, explicit stopping conditions

❌ **Mistake:** Large context window


✅ **Fix:** Implement memory management and summarization

❌ **Mistake:** No logging
✅ **Fix:** Log every LLM call and tool execution

❌ **Mistake:** No cost tracking


✅ **Fix:** Track tokens and API costs from day one

---

## Quick Implementation Checklist

- [ ] Define clear system prompt


- [ ] List all required tools with good descriptions
- [ ] Implement function calling loop
- [ ] Add error handling for tools
- [ ] Set max iterations limit
- [ ] Implement structured output validation
- [ ] Add logging for all operations
- [ ] Set up cost tracking
- [ ] Create test cases
- [ ] Implement observability
- [ ] Add human override capability
- [ ] Deploy with monitoring

---

## Resources

- **Official Docs:** OpenAI API, Anthropic API


- **Example Code:** GitHub repositories for frameworks
- **Best Practices:** AI Engineering books (see earlier document)
- **Tools:** LangSmith, OpenTelemetry, Arize

You might also like