Triển khai multi-turn loop — Full code

Full implementation

Expected output

import json
from typing import Callable
from anthropic import Anthropic
from anthropic.types import Message

client = Anthropic()
MODEL = "claude-sonnet-5-20260205"


# === Tool registry ===
TOOL_FUNCTIONS: dict[str, Callable] = {}

def register_tool(name: str):
    """Decorator to register tool."""
    def decorator(func):
        TOOL_FUNCTIONS[name] = func
        return func
    return decorator


@register_tool("get_current_datetime")
def get_current_datetime(date_format="%Y-%m-%d %H:%M:%S"):
    from datetime import datetime
    if not date_format:
        raise ValueError("date_format cannot be empty")
    return datetime.now().strftime(date_format)


@register_tool("add_duration_to_datetime")
def add_duration_to_datetime(datetime_str, duration, unit="days"):
    from datetime import datetime, timedelta
    dt = datetime.fromisoformat(datetime_str)
    delta = timedelta(**{unit: duration})
    return (dt + delta).isoformat()


@register_tool("set_reminder")
def set_reminder(reminder_text, remind_at_iso):
    from datetime import datetime
    if not reminder_text.strip():
        raise ValueError("Reminder text empty")
    dt = datetime.fromisoformat(remind_at_iso)
    # Simplified: just return confirmation
    return f"Reminder set: '{reminder_text}' at {dt.strftime('%Y-%m-%d %H:%M')}"


# === Schemas ===
get_current_datetime_schema = {
    "name": "get_current_datetime",
    "description": "Get current date/time in strftime format. Use for 'what time is it', 'today's date'.",
    "input_schema": {
        "type": "object",
        "properties": {
            "date_format": {
                "type": "string",
                "description": "Python strftime format",
                "default": "%Y-%m-%d %H:%M:%S"
            }
        },
        "required": []
    }
}

add_duration_to_datetime_schema = {
    "name": "add_duration_to_datetime",
    "description": "Add duration to a datetime. Use when calculating future/past dates.",
    "input_schema": {
        "type": "object",
        "properties": {
            "datetime_str": {"type": "string", "description": "ISO format datetime"},
            "duration": {"type": "integer", "description": "Amount to add"},
            "unit": {"type": "string", "enum": ["seconds", "minutes", "hours", "days", "weeks"]}
        },
        "required": ["datetime_str", "duration", "unit"]
    }
}

set_reminder_schema = {
    "name": "set_reminder",
    "description": "Set a reminder for a future datetime. Use when user asks to set/schedule reminder.",
    "input_schema": {
        "type": "object",
        "properties": {
            "reminder_text": {"type": "string", "description": "What to remind"},
            "remind_at_iso": {"type": "string", "description": "When in ISO format"}
        },
        "required": ["reminder_text", "remind_at_iso"]
    }
}


# === Helpers ===
def add_user_message(messages, content):
    if isinstance(content, Message):
        content = content.content
    messages.append({"role": "user", "content": content})


def add_assistant_message(messages, content):
    if isinstance(content, Message):
        content = content.content
    messages.append({"role": "assistant", "content": content})


def text_from_message(message):
    return "\n".join(b.text for b in message.content if b.type == "text")


def chat(messages, tools=None, system=None, temperature=1.0):
    params = {
        "model": MODEL, "max_tokens": 1000,
        "messages": messages, "temperature": temperature
    }
    if tools:
        params["tools"] = tools
    if system:
        params["system"] = system
    return client.messages.create(**params)


# === Tool runner ===
def run_tool(name: str, input_data: dict) -> str:
    if name not in TOOL_FUNCTIONS:
        raise ValueError(f"Unknown tool: {name}")
    return TOOL_FUNCTIONS[name](**input_data)


def run_tools_for_response(response) -> list:
    tool_uses = [b for b in response.content if b.type == "tool_use"]
    results = []
    for tu in tool_uses:
        try:
            output = run_tool(tu.name, tu.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": tu.id,
                "content": str(output) if not isinstance(output, (dict, list)) else json.dumps(output),
                "is_error": False
            })
        except Exception as e:
            results.append({
                "type": "tool_result",
                "tool_use_id": tu.id,
                "content": f"Error: {str(e)}",
                "is_error": True
            })
    return results


# === Main loop ===
def run_conversation(
    initial_user_msg: str,
    tools: list,
    system: str = None,
    max_turns: int = 10,
    verbose: bool = True
) -> tuple[str, list]:
    """Run multi-turn conversation with tools.
    
    Returns: (final_text, messages_history)
    """
    messages = []
    add_user_message(messages, initial_user_msg)
    
    for turn in range(max_turns):
        if verbose:
            print(f"\n--- Turn {turn + 1} ---")
        
        response = chat(messages, tools=tools, system=system)
        add_assistant_message(messages, response)
        
        # Show any text
        text = text_from_message(response)
        if verbose and text:
            print(f"Claude: {text}")
        
        # End?
        if response.stop_reason == "end_turn":
            if verbose:
                print(f"✓ Done after {turn + 1} turns")
            return text, messages
        
        # Tool use?
        if response.stop_reason == "tool_use":
            tool_uses = [b for b in response.content if b.type == "tool_use"]
            for tu in tool_uses:
                if verbose:
                    print(f"→ Tool: {tu.name}({json.dumps(tu.input)})")
            
            results = run_tools_for_response(response)
            add_user_message(messages, results)
            
            for r in results:
                icon = "✗" if r["is_error"] else "✓"
                if verbose:
                    print(f"  {icon} Result: {r['content'][:100]}")
            continue
        
        # Other stop reasons
        raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")
    
    raise RuntimeError(f"Exceeded max_turns={max_turns}. Possible loop.")


# === Run ===
if __name__ == "__main__":
    final_text, history = run_conversation(
        "Set a reminder for my doctor's appointment, which is 177 days after Jan 1st, 2050.",
        tools=[
            get_current_datetime_schema,
            add_duration_to_datetime_schema,
            set_reminder_schema
        ]
    )
    
    print(f"\n=== Final answer ===\n{final_text}")

Expected output

--- Turn 1 ---
Claude: I'll calculate the date and set the reminder.
→ Tool: add_duration_to_datetime({"datetime_str": "2050-01-01T00:00:00", "duration": 177, "unit": "days"})
  ✓ Result: 2050-06-27T00:00:00

--- Turn 2 ---
→ Tool: set_reminder({"reminder_text": "Doctor's appointment", "remind_at_iso": "2050-06-27T00:00:00"})
  ✓ Result: Reminder set: 'Doctor's appointment' at 2050-06-27 00:00

--- Turn 3 ---
Claude: I've set a reminder for your doctor's appointment on June 27, 2050.
✓ Done after 3 turns

=== Final answer ===
I've set a reminder for your doctor's appointment on June 27, 2050.

Observability

Production cần log:

Dashboard từ log:

• Avg turns per conversation
• Tool success rate
• Cost per conversation
• P99 latency

import time
import logging

logger = logging.getLogger(__name__)


def run_conversation_observed(initial_user_msg, tools, **kwargs):
    start = time.time()
    total_tokens = {"input": 0, "output": 0}
    tool_call_count = 0
    
    # ... (same loop structure)
    
    for turn in range(max_turns):
        response = chat(messages, tools=tools)
        
        # Log usage
        total_tokens["input"] += response.usage.input_tokens
        total_tokens["output"] += response.usage.output_tokens
        
        logger.info(f"Turn {turn+1}: "
                    f"in={response.usage.input_tokens} "
                    f"out={response.usage.output_tokens} "
                    f"stop={response.stop_reason}")
        
        # ... handle tool calls
        if tool_uses:
            tool_call_count += len(tool_uses)
    
    elapsed = time.time() - start
    logger.info(f"Conversation done: "
                f"turns={turn+1} tools={tool_call_count} "
                f"time={elapsed:.1f}s tokens={total_tokens}")

Error patterns

Pattern: Tool retry

Claude error → retry với params sửa. OK.

Bug: Claude lặp lại cùng tool call cùng params → infinite loop.

Detect:

Pattern: Token budget

Conversation dài → input tokens tăng.

recent_tool_signatures = []

for turn in range(max_turns):
    ...
    for tu in tool_uses:
        signature = f"{tu.name}:{json.dumps(tu.input, sort_keys=True)}"
        if signature in recent_tool_signatures[-3:]:
            logger.warning(f"Loop detected: {signature}")
            # Force exit or return error

Pattern: Token budget

if total_tokens["input"] > 100_000:
    raise RuntimeError("Token budget exceeded")

Anti-patterns

❌ Không limit max_turns

Claude stuck → infinite loop, burn money.

Fix: Always max_turns=10-50.

❌ Không log

Khi bug, không biết turn nào crash.

Fix: Structured log mỗi turn.

❌ Silent tool failures

Tool raise exception → pipeline crash vs gracefully handle.

Fix: Try-except quanh run_tool, pass error to Claude.

❌ Hardcoded TOOL_FUNCTIONS dict

Adding new tool cần 2 chỗ (func + dict).

Fix: Dùng @register_tool decorator.

Áp dụng ngay

Bài tập 1: Run production loop (30 phút)

Copy code full ở trên. Test với:

Quan sát turn pattern. Tất cả đều success?

Bài tập 2: Add observability (20 phút)

Extend code để log:

• "Remind me to call mom in 3 hours"
• "What's 42 days before next Tuesday?"
• "Schedule 3 reminders: 1 day, 1 week, 1 month"
• Token count per turn
• Tool latency
• Final cost estimate

Tóm tắt

🎯 Production loop cần: max_turns, logging, tool registry, error handling.

🎯 @register_tool decorator tránh maintain 2 dict.

🎯 Check stop_reason mỗi turn — end_turn vs tool_use vs max_tokens.

🎯 Log usage, tool calls, latency cho observability.

🎯 Detect infinite loops qua signature tracking.