· 20 分で読める · 10,043 文字
Learn how to build persistent AI agents using Claude with Cowork, enabling continuous task automation that survives across sessions and executes complex workflows without manual intervention.
Building Persistent AI Agents with Claude Cowork
Understanding Persistent Agents in Claude Cowork
Persistent agents represent a significant evolution in AI automation. Unlike traditional single-turn interactions with language models, persistent agents in Claude Cowork maintain state across multiple sessions, allowing them to autonomously manage long-running tasks, handle interruptions, and resume work without losing context.
A persistent agent functions as a continuously running process that can be paused, resumed, and monitored. This is particularly valuable for workflows that span hours or days, such as data processing pipelines, monitoring systems, or complex research tasks.
Core Concepts and Architecture
What Makes an Agent "Persistent"
Persistence in agent design means the agent maintains three critical elements:
- State Management: The agent remembers previous decisions, completed tasks, and current progress
- Session Independence: The agent continues functioning even if your application restarts or the connection drops
- Autonomous Execution: The agent can operate without real-time human supervision, making decisions within defined parameters
How Cowork Enables Persistence
Cowork is Claude's framework for building multi-agent systems with built-in coordination capabilities. It handles the complexity of maintaining agent state, managing communication between agents, and ensuring consistent execution across distributed systems.
Setting Up Your First Persistent Agent
Prerequisites
Before building your persistent agent, ensure you have:
- Claude API access (tested with Claude 3.5 Sonnet model, January 2025)
- Python 3.9 or higher
- The Anthropic Python SDK installed
- A basic understanding of async/await patterns in Python
Installation and Initial Setup
Let's start by installing the required packages and setting up your environment:
# Install the Anthropic SDK with Cowork support
pip install anthropic --upgrade
# Verify installation
python -c "import anthropic; print(anthropic.__version__)"
Create a configuration file to manage your API credentials securely:
import os
from anthropic import Anthropic
# Initialize the Anthropic client
# The API key is automatically read from ANTHROPIC_API_KEY environment variable
client = Anthropic()
# Test the connection
def test_connection():
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=100,
messages=[
{"role": "user", "content": "Say 'Connection successful' if you can read this."}
]
)
print(response.content[0].text)
if __name__ == "__main__":
test_connection()
Building a Practical Persistent Agent Example
Creating an Agent with Task Memory
Let's build a practical data processing agent that maintains state across multiple runs. This agent will track completed tasks, handle errors, and resume from where it stopped:
import json
import os
from datetime import datetime
from anthropic import Anthropic
class PersistentDataAgent:
def __init__(self, state_file="agent_state.json"):
self.client = Anthropic()
self.state_file = state_file
self.model = "claude-3-5-sonnet-20241022"
self.conversation_history = []
self.agent_state = self.load_state()
def load_state(self):
"""Load agent state from persistent storage"""
if os.path.exists(self.state_file):
with open(self.state_file, 'r') as f:
return json.load(f)
return {
"tasks_completed": 0,
"current_task": None,
"task_history": [],
"last_execution": None,
"errors": []
}
def save_state(self):
"""Persist agent state to disk"""
self.agent_state["last_execution"] = datetime.now().isoformat()
with open(self.state_file, 'w') as f:
json.dump(self.agent_state, f, indent=2)
def process_task(self, task_description):
"""Process a task while maintaining conversation context"""
# Add context about previous work
system_prompt = f"""You are a persistent AI agent helping with data processing tasks.
Previous work completed: {self.agent_state['tasks_completed']} tasks
Current task: {self.agent_state['current_task']}
Previous errors (if any): {', '.join(self.agent_state['errors'][-3:]) if self.agent_state['errors'] else 'None'}
When processing tasks:
1. Be specific about what you're doing
2. Report progress clearly
3. List any issues encountered
4. Suggest next steps
Remember: You're resuming a potentially interrupted workflow. Check all previous context."""
# Add user message
self.conversation_history.append({
"role": "user",
"content": task_description
})
# Get response from Claude
response = self.client.messages.create(
model=self.model,
max_tokens=1500,
system=system_prompt,
messages=self.conversation_history
)
assistant_message = response.content[0].text
# Store in conversation history
self.conversation_history.append({
"role": "assistant",
"content": assistant_message
})
# Update agent state
self.agent_state["tasks_completed"] += 1
self.agent_state["current_task"] = task_description[:100]
self.agent_state["task_history"].append({
"task": task_description,
"timestamp": datetime.now().isoformat(),
"status": "completed"
})
self.save_state()
return assistant_message
def get_status(self):
"""Report the agent's current status"""
return {
"tasks_completed": self.agent_state["tasks_completed"],
"task_history": self.agent_state["task_history"][-5:], # Last 5 tasks
"last_execution": self.agent_state["last_execution"],
"conversation_turns": len(self.conversation_history)
}
# Example usage
if __name__ == "__main__":
agent = PersistentDataAgent()
# First task
result1 = agent.process_task("Analyze this dataset: [1,2,3,4,5]. Calculate the mean.")
print("First task result:")
print(result1)
print("\n" + "="*50 + "\n")
# Second task - agent remembers context
result2 = agent.process_task("Now calculate the standard deviation and compare it with the mean.")
print("Second task result:")
print(result2)
# Check status
print("\nAgent status:")
print(json.dumps(agent.get_status(), indent=2))
Advanced Patterns for Multi-Session Persistence
Handling Agent Recovery and Resumption
One critical aspect of persistent agents is gracefully handling interruptions. Here's how to implement recovery logic:
import time
from typing import Optional
class ResilientPersistentAgent:
def __init__(self, state_file="agent_state.json", max_retries=3):
self.client = Anthropic()
self.state_file = state_file
self.max_retries = max_retries
self.model = "claude-3-5-sonnet-20241022"
self.agent_state = self.load_state()
def load_state(self):
"""Load state with validation"""
if not os.path.exists(self.state_file):
return self.create_initial_state()
try:
with open(self.state_file, 'r') as f:
state = json.load(f)
# Validate state structure
required_keys = ["tasks_completed", "current_task", "in_progress_task"]
if all(key in state for key in required_keys):
return state
except (json.JSONDecodeError, IOError) as e:
print(f"Warning: Could not load state file: {e}")
return self.create_initial_state()
def create_initial_state(self):
"""Create fresh state structure"""
return {
"tasks_completed": 0,
"current_task": None,
"in_progress_task": None,
"failed_tasks": [],
"session_count": 0,
"creation_time": datetime.now().isoformat()
}
def resume_interrupted_task(self) -> Optional[str]:
"""Check if there's an unfinished task and attempt to resume"""
if self.agent_state["in_progress_task"]:
print("Found interrupted task. Attempting to resume...")
task_data = self.agent_state["in_progress_task"]
# Build resumption prompt
resume_prompt = f"""
I was working on this task before being interrupted:
Task: {task_data['task']}
Previous progress: {task_data['progress']}
Please help me continue from where I left off. What should be the next steps?
"""
return resume_prompt
return None
def execute_with_retry(self, task_description: str, retry_count: int = 0) -> bool:
"""Execute task with automatic retry logic"""
try:
# Mark task as in progress
self.agent_state["in_progress_task"] = {
"task": task_description,
"progress": "started",
"attempt": retry_count + 1
}
self.save_state()
# Execute the task
response = self.client.messages.create(
model=self.model,
max_tokens=2000,
messages=[{
"role": "user",
"content": task_description
}]
)
# Task completed successfully
self.agent_state["tasks_completed"] += 1
self.agent_state["in_progress_task"] = None
self.save_state()
return True
except Exception as e:
print(f"Task execution failed (attempt {retry_count + 1}): {e}")
if retry_count < self.max_retries:
# Wait before retrying (exponential backoff)
wait_time = 2 ** retry_count
print(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
return self.execute_with_retry(task_description, retry_count + 1)
else:
# Record failed task
self.agent_state["failed_tasks"].append({
"task": task_description,
"error": str(e),
"timestamp": datetime.now().isoformat()
})
self.save_state()
return False
def save_state(self):
"""Save state with safety checks"""
self.agent_state["session_count"] = self.agent_state.get("session_count", 0) + 1
with open(self.state_file, 'w') as f:
json.dump(self.agent_state, f, indent=2)
Common Pitfalls and Solutions
State Corruption and Consistency Issues
Problem: Concurrent writes to the state file can corrupt data, especially if multiple agent instances run simultaneously.
Solution: Implement file locking mechanisms:
import fcntl
def save_state_safely(state_dict, filename):
"""Save state with file-level locking"""
# Write to temporary file first
temp_file = filename + ".tmp"
with open(temp_file, 'w') as f:
# Lock the temporary file
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
try:
json.dump(state_dict, f, indent=2)
f.flush()
os.fsync(f.fileno())
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
# Atomic rename
os.replace(temp_file, filename)
Token Limit Overflow in Long Conversations
Problem: Persistent agents maintaining long conversation histories will eventually exceed Claude's token limit (200K tokens for Claude 3.5 Sonnet).
Solution: Implement conversation summarization:
def summarize_old_conversations(conversation_history, max_turns=20):
"""Keep only recent turns and summarize older ones"""
if len(conversation_history) <= max_turns:
return conversation_history
# Keep the last max_turns/2 exchanges
recent_messages = conversation_history[-(max_turns // 2):]
# Summarize everything before that
if len(conversation_history) > max_turns:
older_messages = conversation_history[:-(max_turns // 2)]
# Create a summary prompt
summary_response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=500,
messages=[{
"role": "user",
"content": f"Summarize this conversation in 2-3 sentences for context: {json.dumps(older_messages)}"
}]
)
# Return summary as system context + recent messages
summary_message = {
"role": "user",
"content": f"[Summary of earlier conversation: {summary_response.content[0].text}]\n\nNow continuing with recent work..."
}
return [summary_message] + recent_messages
return conversation_history
Agent Drift and Inconsistent Behavior
Problem: Over time, agents may deviate from their original instructions as they adapt to accumulated context.
Solution: Use explicit system prompts that reset per session:
def execute_task_with_stable_behavior(task, agent_state):
"""Execute task with consistent system prompt"""
# Always include core instructions, don't rely on conversation history
core_system_prompt = """You are a data processing agent. Your responsibilities:
1. Process tasks accurately and report progress
2. Follow the original instructions, not adaptations from past conversations
3. Flag any ambiguities or conflicts
4. Maintain consistent formatting in outputs
Current session information:
- Tasks completed this session: [TASK_COUNT]
- Agent role: Data Processor
- Operating mode: Persistent with state management
"""
final_prompt = core_system_prompt.replace(
"[TASK_COUNT]",
str(agent_state.get("tasks_completed", 0))
)
return client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1500,
system=final_prompt,
messages=[{"role": "user", "content": task}]
)