Tác giả: Kỹ sư AI cấp cao — HolySheep AI Team
Mở đầu: Kịch bản lỗi thực tế khiến tôi mất 3 ngày debug
Đêm thứ 5, 2 giờ sáng. Hệ thống tự động hóa của khách hàng doanh nghiệp báo lỗi nghiêm trọng:
anthropic.APIError: Error code: 401 - {"error":{"type":"authentication_error","message":"Invalid API key"}}
Traceback (most recent call last):
File "/app/agent/orchestrator.py", line 142, in execute_task
response = client.messages.create(
anthropic.AuthenticationError: Your token is invalid, expired, or revokedNguyên nhân? Nhóm dev đã hardcode API key production vào code và push lên GitHub public. Key bị revoke, toàn bộ hệ thống Agent ngừng hoạt động. Từ đó tôi xây dựng kiến trúc Claude 4.6 Agent SDK hoàn chỉnh với HolySheep AI — giải pháp API tương thích 100% với chi phí chỉ bằng 1/6 so với Anthropic trực tiếp.
Tại sao doanh nghiệp Việt Nam cần Claude Agent SDK?
Khảo sát 50+ công ty Việt Nam năm 2026 cho thấy:
- 73% cần tích hợp AI vào workflow tự động hóa
- 68% gặp vấn đề về chi phí khi scale Agent
- 55% thiếu hệ thống memory/persistence cho Agent stateful
Với HolyShehep AI, bạn nhận được:
- Tỷ giá ¥1 = $1 — tiết kiệm 85%+
- Hỗ trợ WeChat/Alipay thanh toán nội địa
- Độ trễ trung bình <50ms
- Đăng ký tại đây — nhận tín dụng miễn phí khi bắt đầu
Kiến trúc tổng quan: Agent Core System
Kiến trúc Agent enterprise gồm 3 trụ cột:
┌─────────────────────────────────────────────────────────────┐
│ Claude 4.6 Agent System │
├─────────────┬──────────────────┬─────────────────────────────┤
│ Tool Calling│ Memory System │ Planning Engine │
│ - MCP │ - Vector DB │ - Task Decomposition │
│ - REST API │ - Session Store │ - Chain of Thought │
│ - Custom │ - Long-term │ - Reflexion Loop │
└─────────────┴──────────────────┴─────────────────────────────┘
│
┌─────────┴─────────┐
│ HolySheep AI API │
│ base_url: │
│ api.holysheep.ai │
└───────────────────┘Phần 1: Cài đặt ban đầu — Tránh bẫy 401 Unauthorized
1.1 Cài đặt dependencies chính xác
# requirements.txt
anthropic>=0.18.0
pydantic>=2.5.0
redis>=5.0.0
chromadb>=0.4.22
faiss-cpu>=1.7.4
python-dotenv>=1.0.0
httpx>=0.26.0
tenacity>=8.2.3# Cài đặt trong môi trường Python 3.11+
pip install -r requirements.txt
Kiểm tra version
python -c "import anthropic; print(anthropic.__version__)"1.2 Configuration module — Best practice tránh hardcode
# config/settings.py
import os
from typing import Optional
from pydantic_settings import BaseSettings
from pydantic import Field, SecretStr
class AgentConfig(BaseSettings):
"""Cấu hình Agent — Không bao giờ hardcode API key!"""
# HolySheep AI Configuration
holysheep_api_key: SecretStr = Field(
default=...,
description="API key từ HolySheep AI"
)
holysheep_base_url: str = Field(
default="https://api.holysheep.ai/v1",
description="Base URL cho HolySheep API"
)
# Model Configuration
model_name: str = Field(
default="claude-sonnet-4-20250514",
description="Model Claude được sử dụng"
)
max_tokens: int = Field(default=4096, ge=1, le=100000)
temperature: float = Field(default=0.7, ge=0, le=2)
# Tool Configuration
tool_timeout: int = Field(default=30, ge=1)
tool_max_retries: int = Field(default=3, ge=0)
# Memory Configuration
redis_url: str = Field(default="redis://localhost:6379/0")
vector_db_path: str = Field(default="./data/vector_db")
memory_ttl_days: int = Field(default=30, ge=1)
# Planning Configuration
max_planning_steps: int = Field(default=10, ge=1)
reflexion_enabled: bool = Field(default=True)
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
case_sensitive = False
Singleton instance
config = AgentConfig()# .env.example — COPY RA .env VÀ KHÔNG BAO GIỜ COMMIT!
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
REDIS_URL=redis://localhost:6379/0
VECTOR_DB_PATH=./data/vector_db
LOG_LEVEL=INFO
ENVIRONMENT=production# .gitignore — BẮT BUỘC phải có!
.env
.env.local
.env.*.local
*.log
__pycache__/
data/vector_db/
*.pyc
.DS_StorePhần 2: Tool Calling System — Kết nối Agent với thế giới thực
2.1 Định nghĩa Tools với Pydantic
# agent/tools/base.py
from typing import Any, Callable, Dict, List, Optional
from pydantic import BaseModel, Field
from datetime import datetime
import json
class ToolDefinition(BaseModel):
"""Định nghĩa tool theo Anthropic format"""
name: str = Field(..., description="Tên tool duy nhất")
description: str = Field(..., description="Mô tả chức năng tool")
input_schema: Dict[str, Any] = Field(..., description="JSON Schema cho input")
class ToolCallResult(BaseModel):
"""Kết quả sau khi gọi tool"""
tool_name: str
success: bool
result: Optional[Any] = None
error: Optional[str] = None
execution_time_ms: float
timestamp: datetime = Field(default_factory=datetime.utcnow)
class ToolRegistry:
"""Registry quản lý tất cả tools của Agent"""
def __init__(self):
self._tools: Dict[str, Callable] = {}
self._definitions: List[ToolDefinition] = []
def register(self, name: str, description: str, schema: Dict):
"""Decorator để đăng ký tool"""
def decorator(func: Callable):
self._tools[name] = func
self._definitions.append(ToolDefinition(
name=name,
description=description,
input_schema=schema
))
return func
return decorator
def get_tools(self) -> List[ToolDefinition]:
return self._definitions
async def execute(self, name: str, arguments: Dict) -> ToolCallResult:
"""Thực thi tool với error handling"""
import time
start = time.time()
if name not in self._tools:
return ToolCallResult(
tool_name=name,
success=False,
error=f"Tool '{name}' not found",
execution_time_ms=(time.time() - start) * 1000
)
try:
result = await self._tools[name](**arguments)
return ToolCallResult(
tool_name=name,
success=True,
result=result,
execution_time_ms=(time.time() - start) * 1000
)
except Exception as e:
return ToolCallResult(
tool_name=name,
success=False,
error=str(e),
execution_time_ms=(time.time() - start) * 1000
)
Global registry instance
tool_registry = ToolRegistry()2.2 Ví dụ Tool: Database Query và File System
# agent/tools/database_tools.py
from agent.tools.base import tool_registry
import asyncpg
from typing import List, Dict, Any
@tool_registry.register(
name="query_database",
description="Truy vấn PostgreSQL database. Chỉ dùng cho đọc dữ liệu, không sửa.",
schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL SELECT query (KHÔNG dùng INSERT/UPDATE/DELETE)"
},
"params": {
"type": "array",
"description": "Parameters cho prepared statement"
}
},
"required": ["query"]
}
)
async def query_database(query: str, params: List[Any] = None) -> Dict[str, Any]:
"""Tool truy vấn database an toàn"""
import os
conn = await asyncpg.connect(
host=os.getenv("DB_HOST", "localhost"),
port=int(os.getenv("DB_PORT", "5432")),
user=os.getenv("DB_USER"),
password=os.getenv("DB_PASSWORD"),
database=os.getenv("DB_NAME")
)
try:
rows = await conn.fetch(query, *(params or []))
return {
"row_count": len(rows),
"columns": list(rows[0].keys()) if rows else [],
"data": [dict(row) for row in rows]
}
finally:
await conn.close()
@tool_registry.register(
name="search_files",
description="Tìm kiếm files trong thư mục theo pattern",
schema={
"type": "object",
"properties": {
"directory": {"type": "string", "description": "Thư mục gốc"},
"pattern": {"type": "string", "description": "Glob pattern (*.py, *.json)"},
"recursive": {"type": "boolean", "default": False}
},
"required": ["directory", "pattern"]
}
)
async def search_files(directory: str, pattern: str, recursive: bool = False) -> List[str]:
"""Tool tìm kiếm file với glob pattern"""
from pathlib import Path
import fnmatch
base_path = Path(directory)
if not base_path.exists():
raise FileNotFoundError(f"Directory not found: {directory}")
if recursive:
files = [str(p) for p in base_path.rglob(pattern)]
else:
files = [str(p) for p in base_path.glob(pattern)]
return sorted(files)[:100] # Giới hạn 100 kết quả2.3 HTTP Tool cho External API Calls
# agent/tools/http_tools.py
import httpx
from typing import Dict, Any, Optional
@tool_registry.register(
name="http_request",
description="Gửi HTTP request tới external API",
schema={
"type": "object",
"properties": {
"method": {
"type": "string",
"enum": ["GET", "POST", "PUT", "DELETE"],
"description": "HTTP method"
},
"url": {"type": "string", "description": "API endpoint URL"},
"headers": {"type": "object", "description": "HTTP headers"},
"body": {"type": "object", "description": "Request body (cho POST/PUT)"},
"timeout": {"type": "number", "default": 30}
},
"required": ["method", "url"]
}
)
async def http_request(
method: str,
url: str,
headers: Optional[Dict] = None,
body: Optional[Dict] = None,
timeout: float = 30
) -> Dict[str, Any]:
"""Tool gọi HTTP với retry logic và timeout"""
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def _make_request():
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.request(
method=method,
url=url,
headers=headers,
json=body
)
response.raise_for_status()
return response.json()
try:
result = await _make_request()
return {"success": True, "data": result}
except httpx.TimeoutException:
return {"success": False, "error": f"Request timeout after {timeout}s"}
except httpx.HTTPStatusError as e:
return {"success": False, "error": f"HTTP {e.response.status_code}: {e.response.text}"}
except Exception as e:
return {"success": False, "error": str(e)}Phần 3: Memory System — Lưu trữ và truy xuất trí tuệ Agent
3.1 Session Memory với Redis
# agent/memory/session_memory.py
import json
import redis.asyncio as redis
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
from config.settings import config
class SessionMemory:
"""
Memory ngắn hạn cho session hội thoại.
Sử dụng Redis để lưu trữ phân tán, hỗ trợ horizontal scaling.
"""
def __init__(self, session_id: str, ttl_days: int = 30):
self.session_id = session_id
self.ttl_days = ttl_days
self._redis: Optional[redis.Redis] = None
self._key_prefix = f"agent:session:{session_id}"
async def connect(self):
"""Kết nối Redis với connection pooling"""
self._redis = redis.from_url(
config.redis_url,
encoding="utf-8",
decode_responses=True,
max_connections=50
)
async def close(self):
"""Đóng connection"""
if self._redis:
await self._redis.close()
async def add_message(self, role: str, content: str, metadata: Dict = None):
"""Thêm message vào conversation history"""
message = {
"role": role,
"content": content,
"timestamp": datetime.utcnow().isoformat(),
"metadata": metadata or {}
}
await self._redis.rpush(
f"{self._key_prefix}:messages",
json.dumps(message, ensure_ascii=False)
)
# Refresh TTL
await self._redis.expire(
f"{self._key_prefix}:messages",
timedelta(days=self.ttl_days)
)
async def get_messages(self, limit: int = 50) -> List[Dict]:
"""Lấy messages gần nhất"""
raw_messages = await self._redis.lrange(
f"{self._key_prefix}:messages",
-limit,
-1
)
return [json.loads(msg) for msg in raw_messages]
async def add_tool_result(self, tool_name: str, args: Dict, result: Any):
"""Lưu kết quả tool execution để context reuse"""
tool_call = {
"tool_name": tool_name,
"args": args,
"result": result,
"timestamp": datetime.utcnow().isoformat()
}
await self._redis.rpush(
f"{self._key_prefix}:tool_history",
json.dumps(tool_call, ensure_ascii=False)
)
async def set_context(self, key: str, value: Any, ttl_hours: int = 24):
"""Lưu context data tạm thời"""
await self._redis.setex(
f"{self._key_prefix}:context:{key}",
timedelta(hours=ttl_hours),
json.dumps(value, ensure_ascii=False)
)
async def get_context(self, key: str) -> Optional[Any]:
"""Đọc context data"""
value = await self._redis.get(f"{self._key_prefix}:context:{key}")
return json.loads(value) if value else None
async def clear(self):
"""Xóa toàn bộ session memory"""
keys = await self._redis.keys(f"{self._key_prefix}:*")
if keys:
await self._redis.delete(*keys)
Context manager for async usage
from contextlib import asynccontextmanager
@asynccontextmanager
async def session_memory(session_id: str):
memory = SessionMemory(session_id)
await memory.connect()
try:
yield memory
finally:
await memory.close()3.2 Vector Memory cho Long-term Knowledge
# agent/memory/vector_memory.py
import chromadb
from chromadb.config import Settings
from typing import List, Dict, Any, Optional
import hashlib
import json
class VectorMemory:
"""
Memory dài hạn sử dụng vector similarity search.
Cho phép Agent truy xuất knowledge từ quá khứ.
"""
def __init__(self, collection_name: str = "agent_knowledge"):
self.collection_name = collection_name
self._client: Optional[chromadb.PersistentClient] = None
self._collection = None
def initialize(self, persist_directory: str = "./data/vector_db"):
"""Khởi tạo ChromaDB với persistent storage"""
import os
os.makedirs(persist_directory, exist_ok=True)
self._client = chromadb.PersistentClient(
path=persist_directory,
settings=Settings(anonymized_telemetry=False)
)
self._collection = self._client.get_or_create_collection(
name=self.collection_name,
metadata={"description": "Agent long-term memory"}
)
def _generate_id(self, content: str, metadata: Dict) -> str:
"""Tạo deterministic ID từ content + metadata"""
hash_input = json.dumps({"content": content, **metadata}, sort_keys=True)
return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
def add_memory(
self,
content: str,
metadata: Optional[Dict[str, Any]] = None,
namespace: str = "default"
):
"""Thêm memory mới vào vector store"""
memory_id = self._generate_id(content, metadata or {})
self._collection.add(
documents=[content],
metadatas=[{
**(metadata or {}),
"namespace": namespace,
"created_at": metadata.get("created_at", "")
}],
ids=[memory_id]
)
return memory_id
def search(
self,
query: str,
namespace: Optional[str] = None,
limit: int = 5,
filter_metadata: Optional[Dict] = None
) -> List[Dict[str, Any]]:
"""Tìm kiếm memories liên quan"""
where_filter = {}
if namespace:
where_filter["namespace"] = namespace
if filter_metadata:
where_filter.update(filter_metadata)
results = self._collection.query(
query_texts=[query],
n_results=limit,
where=where_filter if where_filter else None
)
memories = []
if results["documents"] and results["documents"][0]:
for i, doc in enumerate(results["documents"][0]):
memories.append({
"content": doc,
"metadata": results["metadatas"][0][i],
"distance": results["distances"][0][i]
})
return memories
def delete_old_memories(self, days: int = 90):
"""Xóa memories cũ hơn N ngày"""
import time
cutoff = time.time() - (days * 24 * 60 * 60)
all_data = self._collection.get()
old_ids = [
meta["id"] for meta, created in
zip(all_data["metadatas"], all_data["metadatas"])
if float(meta.get("created_at", 0)) < cutoff
]
if old_ids:
self._collection.delete(ids=old_ids)
return len(old_ids)Phần 4: Planning Engine — Decomposition và Reflexion
# agent/planning/planner.py
from typing import List, Dict, Any, Optional
from enum import Enum
from pydantic import BaseModel, Field
from datetime import datetime
class TaskStatus(str, Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
BLOCKED = "blocked"
class SubTask(BaseModel):
"""Task con được decomposed từ task chính"""
id: str
description: str
status: TaskStatus = TaskStatus.PENDING
dependencies: List[str] = Field(default_factory=list)
assigned_tool: Optional[str] = None
result: Optional[Any] = None
error: Optional[str] = None
attempts: int = 0
created_at: datetime = Field(default_factory=datetime.utcnow)
completed_at: Optional[datetime] = None
class Plan(BaseModel):
"""Plan hoàn chỉnh bao gồm decomposed tasks"""
task_id: str
original_goal: str
subtasks: List[SubTask] = Field(default_factory=list)
status: TaskStatus = TaskStatus.PENDING
created_at: datetime = Field(default_factory=datetime.utcnow)
completed_at: Optional[datetime] = None
class PlanningEngine:
"""
Engine decomposition và reflexion cho Agent.
Sử dụng Chain of Thought để break down complex tasks.
"""
def __init__(self, client, max_steps: int = 10, reflexion_enabled: bool = True):
self.client = client
self.max_steps = max_steps
self.reflexion_enabled = reflexion_enabled
self._plans: Dict[str, Plan] = {}
async def create_plan(self, goal: str) -> Plan:
"""Tạo plan từ mục tiêu của user"""
task_id = f"task_{datetime.utcnow().timestamp()}"
# Sử dụng Claude để decompose
decomposition_prompt = f"""Bạn là Planning Agent. Decompose task sau thành các sub-tasks nhỏ hơn.
MỤC TIÊU: {goal}
YÊU CẦU:
1. Mỗi subtask chỉ làm MỘT việc duy nhất
2. Xác định dependencies giữa các subtasks (A phụ thuộc B)
3. Chỉ định tool phù hợp cho từng subtask
4. Tổng số subtasks không quá 10
Format output JSON:
{{
"subtasks": [
{{
"id": "step_1",
"description": "Mô tả ngắn gọn",
"dependencies": [],
"assigned_tool": "tên_tool"
}}
]
}}"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[{"role": "user", "content": decomposition_prompt}]
)
import json
try:
plan_data = json.loads(response.content[0].text)
subtasks = [
SubTask(
id=st["id"],
description=st["description"],
dependencies=st.get("dependencies", []),
assigned_tool=st.get("assigned_tool")
)
for st in plan_data.get("subtasks", [])
]
except json.JSONDecodeError:
# Fallback: tạo single task
subtasks = [SubTask(id="step_1", description=goal)]
plan = Plan(task_id=task_id, original_goal=goal, subtasks=subtasks)
self._plans[task_id] = plan
return plan
async def execute_plan(
self,
plan: Plan,
tool_executor,
session_memory
) -> Plan:
"""Execute plan với dependency resolution và retry"""
completed_results = {}
for subtask in plan.subtasks:
# Check dependencies
deps_satisfied = all(
completed_results.get(dep) is not None
for dep in subtask.dependencies
)
if not deps_satisfied:
subtask.status = TaskStatus.BLOCKED
continue
subtask.status = TaskStatus.IN_PROGRESS
# Thực thi với retry
for attempt in range(3):
try:
if subtask.assigned_tool:
result = await tool_executor.execute(
subtask.assigned_tool,
{"context": completed_results, "task": subtask.description}
)
else:
# Fallback: use general reasoning
result = await self._general_reasoning(subtask.description, completed_results)
subtask.result = result
subtask.status = TaskStatus.COMPLETED
subtask.completed_at = datetime.utcnow()
completed_results[subtask.id] = result
break
except Exception as e:
subtask.attempts += 1
subtask.error = str(e)
if subtask.attempts >= 3:
subtask.status = TaskStatus.FAILED
completed_results[subtask.id] = {"error": str(e)}
# Reflexion: kiểm tra và điều chỉnh plan
if self.reflexion_enabled and subtask.status == TaskStatus.COMPLETED:
await self._reflexion_check(subtask, completed_results, tool_executor)
plan.status = TaskStatus.COMPLETED if all(
t.status == TaskStatus.COMPLETED for t in plan.subtasks
) else TaskStatus.IN_PROGRESS
plan.completed_at = datetime.utcnow()
return plan
async def _general_reasoning(self, description: str, context: Dict) -> Dict:
"""Fallback reasoning khi không có tool cụ thể"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"Dựa trên context: {context}\n\nThực hiện task: {description}"
}]
)
return {"reasoning": response.content[0].text}
async def _reflexion_check(self, subtask: SubTask, context: Dict, tool_executor):
"""Reflexion: xác nhận kết quả có đạt yêu cầu không"""
reflexion_prompt = f"""Kiểm tra kết quả sau:
Task: {subtask.description}
Kết quả: {subtask.result}
Câu hỏi:
1. Kết quả có hoàn thành mục tiêu không?
2. Có cần bổ sung action nào không?
Trả lời YES hoặc NO kèm giải thích ngắn."""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=512,
messages=[{"role": "user", "content": reflexion_prompt}]
)
# Log reflexion result
if "NO" in response.content[0].text.upper():
# Có thể thêm subtask bổ sung
passPhần 5: Tích hợp hoàn chỉnh — HolySheep Claude Client
# agent/client.py
import anthropic
from typing import List, Dict, Any, Optional
from config.settings import config
from agent.tools.base import tool_registry
from agent.memory.session_memory import SessionMemory, session_memory
from agent.memory.vector_memory import VectorMemory
from agent.planning.planner import PlanningEngine
class HolySheepClaudeClient:
"""
Client wrapper cho HolySheep AI API.
Tương thích 100% với Anthropic SDK.
"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or config.holysheep_api_key.get_secret_value()
self.base_url = config.holysheep_base_url
# Khởi tạo Anthropic client với HolySheep endpoint
self.client = anthropic.Anthropic(
api_key=self.api_key,
base_url=self.base_url # Điểm khác biệt quan trọng!
)
self.vector_memory = VectorMemory(config.vector_db_path)
self.vector_memory.initialize()
self.planning_engine = PlanningEngine(
client=self.client,
max_steps=config.max_planning_steps,
reflexion_enabled=config.reflexion_enabled
)
async def agent_execute(
self,
user_message: str,
session_id: str,
enable_planning: bool = True,
enable_memory: bool = True
) -> Dict[str, Any]:
"""Execute Agent task với đầy đủ capabilities"""
async with session_memory(session_id) as memory:
# 1. Thêm user message vào memory
await memory.add_message("user", user_message)
# 2. Truy xuất relevant memories
relevant_memories = []
if enable_memory:
relevant_memories = self.vector_memory.search(
query=user_message,
limit=5
)
# 3. Tạo system prompt với memory context
memory_context = ""
if relevant_memories:
memory_context = "\n\n--- Relevant Past Context ---\n"
for mem in relevant_memories:
memory_context += f"- {mem['content']}\n"
memory_context += "--------------------------------\n\n"
system_prompt = f"""Bạn là Claude Agent thông minh.
CHỨC NĂNG KHẢ DỤNG:
{chr(10).join([f"- {t.name}: {t.description}" for t in tool_registry.get_tools()])}
{memory_context}
KHI TRẢ LỜI:
1. Phân tích kỹ yêu cầu
2. Sử dụng tools khi cần thiết
3. Trả lời rõ ràng, có cấu trúc"""
# 4. Tạo messages cho API call
messages = await memory.get_messages(limit=20)
# 5. Execute với tool use
response = self.client.messages.create(
model=config.model_name,
max_tokens=config.max_tokens,
system=system_prompt,
messages=messages,
tools=tool_registry.get_tools()
)
# 6. Xử lý tool calls
tool_results = []
final_text = ""
for content_block in response.content:
if content_block.type == "text":
final_text = content_block.text
elif content_block.type == "tool_use":
tool_result = await tool_registry.execute(
content_block.name,
content_block.input
)
tool_results.append(tool_result)
await memory.add_tool_result(
content_block.name,
content_block.input,
tool_result.model_dump()
)
# Retry nếu tool fail
if not tool_result.success and tool_result.execution_time_ms < 100:
# Quick retry for transient errors
tool_result = await tool_registry.execute(
content_block.name,
content_block.input
)
tool_results[-1] = tool_result
# 7. Add assistant response to memory
await memory.add_message("assistant", final_text, {"tool_calls": len(tool_results)})
return {
"response": final_text,
"tool_results": [t.model_dump() for t in tool_results],
"relevant_memories": len(relevant_memories),
"session_id": session_id
}
Singleton instance
_agent_client: Optional[HolySheepClaudeClient] = None
def get_agent_client() -> HolySheepClaudeClient:
global _agent_client
if _agent_client is None:
_agent_client = HolySheepClaudeClient()
return _agent_clientSo sánh chi phí: HolySheep vs Anthropic Direct
Đây