20 KiB
20 KiB
AgentCore.jl - Examples and Patterns
Quick Start Examples
Example 1: Basic Conversation
using AgentCore
# Create model
model = Model(
"gpt-4",
"GPT-4",
"openai",
"openai",
"https://api.openai.com/v1",
true,
["text"],
ModelCost(0.00003, 0.00006, 0.0, 0.0),
128000,
4096,
)
# Create tools
bash_tool = createBashTool()
# Create agent
agent = Agent(Dict(
:systemPrompt => "You are a helpful assistant.",
:model => model,
:tools => [bash_tool],
:thinkingLevel => THINKING_MEDIUM,
:toolExecution => EXECUTION_PARALLEL,
))
# Subscribe to events
subscribe(agent) do event, signal
if event isa MessageEndEvent
println("Agent: $(event.message)")
end
end
# Start conversation
prompt(agent, "What's in the current directory?")
# Wait for completion
wait_for_idle(agent)
# Get final state
state = get_state(agent)
println("Total messages: $(length(state.messages))")
Example 2: Conversation with Memory
# Create session storage
storage = JsonlSessionStorage(
JsonlSessionMetadata(
"session_1",
"2024-01-01T00:00:00Z",
"/path/to/project",
"/path/to/session.jsonl",
nothing,
Dict("project" => "my-project"),
),
"/path/to/session.jsonl",
)
# Create session
session = Session(storage)
# Create agent with session
agent = Agent(Dict(
:systemPrompt => "You are a helpful assistant.",
:model => model,
:tools => [bash_tool],
:sessionId => session.getMetadata().id,
))
# Add messages to session
function addToSession(session, message)
appendMessage(session, message)
end
# Start conversation
prompt(agent, "Hello, my name is Alice.")
# Continue conversation (messages persist in session)
prompt(agent, "What's the weather like today?")
# Check session stats
stats = getSessionStats(session)
println("Messages: $(stats.message_count)")
println("Total tokens: $(stats.total_tokens)")
Example 3: Steering and Follow-Up
# Start conversation
prompt(agent, "Create a Python project.")
# User wants to redirect
steer(agent, UserMessage("user", [TextContent("Actually, let's use Node.js instead")], timestamp))
# Wait for redirection
wait_for_idle(agent)
# Agent would normally stop, but user has more
prompt(agent, "Wait, there's one more thing...")
followUp(agent, UserMessage("user", [TextContent("Can you add tests?")], timestamp))
# Continue until completion
while hasQueuedMessages(agent)
wait_for_idle(agent)
end
Example 4: Branching Conversations
# Initial conversation
prompt(agent, "I want to build a web app.")
# User decides to explore a different path
session.moveTo(msg_3_id) # Go back to message 3
# Create branch
appendBranchSummary(
session,
"User decided to explore mobile app instead",
msg_3_id,
Dict("focus" => "mobile"),
)
# Continue on new branch
prompt(agent, "Let's build a mobile app instead.")
# Check branches
branch = getBranch(session)
println("Current branch has $(length(branch)) entries")
Advanced Patterns
Pattern 1: Long-Running Agent with Compaction
# Configure compaction settings
MAX_TOKENS = 120000 # Stay under 128K limit
COMPACTION_THRESHOLD = 100000
# Agent loop with compaction
function runAgentWithCompaction(agent, session)
while true
# Get current token count
stats = getSessionStats(session)
if stats.total_tokens > COMPACTION_THRESHOLD
# Compact session
compactSession(session)
end
# Check if agent is idle
if !hasQueuedMessages(agent) && !isnothing(agent.active_run)
break
end
end
end
function compactSession(session)
# Get current branch
branch = getBranch(session)
# Calculate tokens to compact
total_tokens = 0
for entry in branch
if entry isa MessageEntry
total_tokens += estimateTokens(entry.message)
end
end
if total_tokens < COMPACTION_THRESHOLD
return
end
# Identify messages to compact
messages_to_compact = []
tokens_to_keep = 50000 # Keep recent 50K tokens
for entry in branch
if entry isa MessageEntry
msg_tokens = estimateTokens(entry.message)
if tokens_to_keep > 0
tokens_to_keep -= msg_tokens
else
push!(messages_to_compact, entry)
end
end
end
# Generate summary
summary = generateSummary(messages_to_compact)
# Create compaction entry
appendCompaction(
session,
summary,
messages_to_compact[end].id,
total_tokens,
)
println("Compacted $(length(messages_to_compact)) messages")
end
function estimateTokens(message::AgentMessage)::Int64
# Simple estimation: ~4 chars per token
content = if message isa UserMessage
join([c.text for c in message.content if c isa TextContent])
elseif message isa AssistantMessage
join([c.text for c in message.content if c isa TextContent])
elseif message isa ToolResultMessage
join([c.text for c in message.content if c isa TextContent])
else
""
end
return ceil(Int, length(content) / 4)
end
function generateSummary(messages::Vector{MessageEntry})::String
# Use LLM to generate summary
summary = "Conversation summary:"
for msg in messages
summary *= "\n- $(msg.message)"
end
return summary
end
Pattern 2: Custom Tool with Context
# Define context type
struct DatabaseContext
connection::Any
user::String
end
# Create tool with context
function createDatabaseTool()
return AgentTool(
"database",
"database",
"Execute SQL queries",
Dict{String, Any}(),
(tool_call_id, params, signal, on_update, context) -> begin
if !isa(context, DatabaseContext)
return AgentToolResult(
[TextContent("Error: Database context not provided")],
nothing,
nothing,
nothing,
true, # terminate
)
end
# Execute query
query = params["query"]
result = executeQuery(context.connection, query)
return AgentToolResult(
[TextContent(formatResult(result))],
Dict("user" => context.user),
nothing,
nothing,
nothing,
)
end,
nothing,
EXECUTION_SEQUENTIAL,
)
end
# Use tool with context
db_context = DatabaseContext(connection, "alice")
harness = AgentHarness(Dict(
:tools => [createDatabaseTool()],
:tool_context => AgentHarnessToolContextSource(db_context),
))
Pattern 3: Dynamic Model Selection
# Hook to change model based on task
function dynamicModelSelection(context, signal)
# Check message content
last_message = context.message
# If complex task, use more capable model
if contains(join(last_message.content), "analyze")
return AgentLoopTurnUpdate(
context = context.context,
model = Model("gpt-4", "GPT-4", "openai", ...),
thinking_level = THINKING_HIGH,
)
end
# Otherwise use cheaper model
return AgentLoopTurnUpdate(
context = context.context,
model = Model("gpt-3.5", "GPT-3.5", "openai", ...),
thinking_level = THINKING_MEDIUM,
)
end
# Configure agent
agent = Agent(Dict(
:prepareNextTurn => dynamicModelSelection,
))
Pattern 4: Rate Limiting
# Rate limiter
struct RateLimiter
calls_per_minute::Int
last_calls::Vector{DateTime}
end
function RateLimiter(calls_per_minute::Int)
return RateLimiter(calls_per_minute, DateTime[])
end
function rateLimit(limiter::RateLimiter)
now = Dates.now()
# Remove old calls
limiter.last_calls = filter(
c -> Dates.value(now - c) / 1000 < 60,
limiter.last_calls,
)
# Check limit
if length(limiter.last_calls) >= limiter.calls_per_minute
return false
end
# Record call
push!(limiter.last_calls, now)
return true
end
# Use in hook
limiter = RateLimiter(60) # 60 calls per minute
function rateLimitHook(event, signal)
if !rateLimit(limiter)
return BeforeProviderPayloadResult(event.payload) # Still send, but track
end
return BeforeProviderPayloadResult(event.payload)
end
# Configure
agent = Agent(Dict(
:beforeProviderPayload => rateLimitHook,
))
Pattern 5: Multi-Step Tool Execution
# Tool that requires multiple steps
function createMultiStepTool()
return AgentTool(
"multistep",
"multistep",
"Multi-step task",
Dict{String, Any}(),
(tool_call_id, params, signal, on_update, context) -> begin
# Step 1: Prepare
on_update("Preparing...")
prepare_result = prepareStep(params)
# Step 2: Execute
on_update("Executing...")
execute_result = executeStep(prepare_result, params)
# Step 3: Finalize
on_update("Finalizing...")
finalize_result = finalizeStep(execute_result)
return AgentToolResult(
[TextContent(finalize_result)],
Dict("steps" => 3),
nothing,
nothing,
nothing,
)
end,
nothing,
EXECUTION_SEQUENTIAL,
)
end
Pattern 6: Image Processing
# Create read tool with image support
image_processor = ReadImageProcessor(
(path, context) -> begin
# Load image
image_data = readImage(path)
# Process with vision model
result = processImageWithVision(image_data)
return ReadImageProcessorResult(
[TextContent(result.description)],
result.usage,
)
end,
context,
)
read_tool = createReadTool(Dict(
"image_processor" => image_processor,
))
Pattern 7: Session Navigation
# Navigate to specific point
session.moveTo(entry_id)
# Get branch from specific point
branch = getBranch(session, entry_id)
# Create label for easy navigation
appendLabel(session, entry_id, "important-decision")
# Find labeled entry
label = getLabel(session, "important-decision")
# Build context from branch
context = buildSessionContext(session)
# Get specific messages
messages = sessionEntryToContextMessages(entry, index, entries)
Pattern 8: Batch Processing
# Process multiple prompts in batch
prompts = [
"What is Julia?",
"What is JavaScript?",
"What is Python?",
]
results = []
for prompt_text in prompts
# Create fresh agent for each prompt
agent = Agent(Dict(
:systemPrompt => "You are a helpful assistant.",
:model => model,
:tools => [bash_tool],
))
# Run prompt
prompt(agent, prompt_text)
wait_for_idle(agent)
# Get result
state = get_state(agent)
last_message = state.messages[end]
push!(results, last_message)
# Clean up
reset!(agent)
end
# Process results
for result in results
println("Result: $(result)")
end
Pattern 9: Custom Event Handling
# Custom event types
struct CustomEvent <: AgentEvent
data::Any
end
# Custom event handler
function customEventHandler(event, signal)
if event isa CustomEvent
println("Custom event: $(event.data)")
end
end
# Subscribe to custom events
subscribe(agent) do event, signal
customEventHandler(event, signal)
end
# Emit custom event
emit(CustomEvent("custom data"))
Pattern 10: Error Handling
# Hook for error handling
function errorHook(context, signal)
if context isa PrepareNextTurnContext
last_message = context.message
if last_message.stop_reason == "error"
println("Error in conversation: $(last_message.error_message)")
return AgentLoopTurnUpdate(
context = context.context,
model = context.context.model,
thinking_level = THINKING_HIGH, # Use more capable model
)
end
end
return nothing
end
# Use in agent
agent = Agent(Dict(
:prepareNextTurn => errorHook,
))
Testing Patterns
Unit Testing
# Test tool execution
@testset "Bash tool" begin
tool = createBashTool()
# Test successful execution
result = tool.execute("tc1", Dict("command" => "echo hello"), nothing, nothing, nothing)
@test result.content[1].text == "hello\n"
@test result.details === nothing
# Test error handling
result = tool.execute("tc2", Dict("command" => "exit 1"), nothing, nothing, nothing)
@test result.terminate === true
end
# Test agent with mock LLM
@testset "Agent with mock" begin
# Mock stream function
function mockStreamFn(model, context, options)
# Return mock response
return MockResponse([TextContent("Hello!")])
end
agent = Agent(Dict(
:stream_fn => mockStreamFn,
:systemPrompt => "You are a helpful assistant.",
:model => model,
))
# Test prompt
prompt(agent, "Hello")
wait_for_idle(agent)
# Verify result
state = get_state(agent)
@test length(state.messages) == 2 # User + Assistant
end
Integration Testing
# Test full conversation flow
@testset "Full conversation" begin
# Create session storage
storage = InMemorySessionStorage(...)
session = Session(storage)
# Create agent
agent = Agent(Dict(
:systemPrompt => "You are a helpful assistant.",
:model => model,
:tools => [bash_tool],
:sessionId => session.getMetadata().id,
))
# Run conversation
prompt(agent, "What's in the directory?")
wait_for_idle(agent)
# Verify session
context = buildSessionContext(session)
@test length(context.messages) == 2
# Continue conversation
prompt(agent, "What's the weather?")
wait_for_idle(agent)
# Verify growth
context = buildSessionContext(session)
@test length(context.messages) == 4
end
Performance Patterns
Pattern 1: Caching
# Simple caching for LLM calls
struct LLMCache
cache::Dict{String, AssistantMessage}
end
function LLMCache()
return LLMCache(Dict{String, AssistantMessage}())
end
function getCached(cache::LLMCache, key::String)
return get(cache.cache, key, nothing)
end
function setCached(cache::LLMCache, key::String, value::AssistantMessage)
cache.cache[key] = value
end
# Use in stream function
function cachedStreamFn(model, context, options)
key = generateCacheKey(context)
cached = getCached(cache, key)
if !isnothing(cached)
return MockResponse(cached)
end
result = actualStreamFn(model, context, options)
setCached(cache, key, result)
return result
end
Pattern 2: Batch LLM Calls
# Batch multiple LLM calls
function batchLLMCalls(calls::Vector{Dict})
results = []
for call in calls
result = streamFunction(
call[:model],
call[:context],
call[:options],
)
push!(results, result)
end
return results
end
# Use with parallel execution
tool.execute = (id, params, signal, on_update, context) -> begin
# Batch multiple LLM calls
llm_calls = [
Dict(:model => model, :context => context1, :options => options1),
Dict(:model => model, :context => context2, :options => options2),
]
results = batchLLMCalls(llm_calls)
return AgentToolResult(
[TextContent(join([r.text for r in results], "\n"))],
nothing,
nothing,
nothing,
nothing,
)
end
Pattern 3: Lazy Loading
# Lazy load skills
struct LazySkills
dir::String
skills::Union{Vector{Skill}, Nothing}
end
function LazySkills(dir)
return LazySkills(dir, nothing)
end
function getSkills(lazy::LazySkills)
if isnothing(lazy.skills)
lazy.skills, _ = loadSkills(lazy.dir)
end
return lazy.skills
end
# Use in harness
harness = AgentHarness(Dict(
:resources => AgentHarnessResources(
templates,
LazySkills("/path/to/skills"),
),
))
Production Patterns
Pattern 1: Observability
# Logging hook
function loggingHook(event, signal)
if event isa BeforeProviderRequestEvent
println("[Request] $(event.model.id)")
elseif event isa AfterProviderResponseEvent
println("[Response] Status: $(event.status)")
elseif event isa ToolExecutionEndEvent
println("[Tool] $(event.tool_name): $(event.is_error ? "error" : "success")")
end
return nothing
end
# Metrics hook
function metricsHook(event, signal)
if event isa AgentStartEvent
metrics.start_time = Dates.now()
elseif event isa AgentEndEvent
duration = Dates.value(Dates.now() - metrics.start_time) / 1000
println("[Metrics] Duration: $(duration)s")
end
return nothing
end
Pattern 2: Retry Logic
# Retry hook
function retryHook(event, signal)
if event isa AfterProviderResponseEvent && event.status >= 500
# Server error, retry
return BeforeProviderRequestResult(Dict(
"retry" => true,
"max_retries" => 3,
))
end
return nothing
end
# Use in stream options
harness = AgentHarness(Dict(
:stream_options => AgentHarnessStreamOptions(
max_retries = 3,
max_retry_delay_ms = 5000,
),
:retry => retryHook,
))
Pattern 3: Security
# Security hook
function securityHook(event, signal)
if event isa ToolCallEvent
# Validate tool call
if event.tool_name == "bash"
command = event.input["command"]
# Block dangerous commands
dangerous_patterns = ["rm -rf /", "sudo", "curl | sh"]
for pattern in dangerous_patterns
if contains(command, pattern)
return ToolCallResult(true, "Blocked dangerous command")
end
end
end
end
return nothing
end
Debugging Patterns
Pattern 1: Conversation Trace
# Trace conversation
trace = []
subscribe(agent) do event, signal
if event isa MessageEndEvent
push!(trace, Dict(
"role" => event.message.role,
"content" => event.message.content,
))
end
end
# Run conversation
prompt(agent, "Hello")
wait_for_idle(agent)
# Print trace
for entry in trace
println("$(entry["role"]): $(entry["content"])")
end
Pattern 2: Tool Call Trace
tool_trace = []
subscribe(agent) do event, signal
if event isa ToolExecutionStartEvent
push!(tool_trace, Dict(
"type" => "start",
"tool" => event.tool_name,
"args" => event.args,
))
elseif event isa ToolExecutionEndEvent
push!(tool_trace, Dict(
"type" => "end",
"tool" => event.tool_name,
"error" => event.is_error,
))
end
end
Pattern 3: State Dump
function dumpState(agent)
state = get_state(agent)
println("=== Agent State ===")
println("System prompt: $(state.system_prompt)")
println("Model: $(state.model.name)")
println("Thinking level: $(state.thinking_level)")
println("Messages: $(length(state.messages))")
println("Tools: $(length(state.tools))")
println("==================")
end
# Use after conversation
prompt(agent, "Hello")
wait_for_idle(agent)
dumpState(agent)
Best Practices Summary
- Start simple, add complexity gradually
- Use hooks for customization, not core logic
- Test with mock LLM first
- Monitor token usage for long conversations
- Use branches for exploration
- Compact periodically to stay within limits
- Handle errors gracefully
- Log important events
- Test edge cases
- Profile performance