第四章：运行时引擎¶

class QueryEngine:
    async def submitMessage(self, user_input: str):
        """异步生成器循环,每次 yield 一个事件"""
        yield AgentStartEvent()

        # 初始化会话状态
        app_state = self.init_app_state(user_input)

        while app_state.has_work_to_do():
            yield TurnStartEvent()

            # 构建完整的消息和上下文
            # 考虑缓存边界,分离系统提示词和动态上下文
            system_prompt = self.get_system_prompt()
            messages = await self.build_messages(app_state)

            # 检查Token预算
            if self.estimate_tokens(system_prompt + messages) > self.token_budget:
                messages = self.auto_compact_history(messages)

            # 流式调用模型
            stream = await self.model_client.create_message_stream(
                messages=messages,
                system=system_prompt,
                tools=self.get_available_tools()
            )

            # 逐个处理流中的事件
            # 按 Anthropic 流式协议增量累积内容块,在消息终止时才执行工具
            assembler = MessageAssembler()
            async for event in stream:
                if event.type == "content_block_start":
                    # 新的内容块开始(文本或工具使用),登记到累积器
                    assembler.start_block(event.index, event.content_block)
                    yield event

                elif event.type == "content_block_delta":
                    # 把增量累积进当前块(文本/工具入参 JSON)
                    assembler.append_delta(event.index, event.delta)
                    if event.delta.type == "text_delta":
                        yield TextDeltaEvent(event.delta.text)
                    elif event.delta.type == "input_json_delta":
                        yield ToolInputDeltaEvent(event.delta.partial_json)

                elif event.type == "content_block_stop":
                    # 当前块完成,定稿(如解析工具入参 JSON)
                    assembler.finish_block(event.index)

                elif event.type == "message_delta":
                    # 顶层增量:仅携带 stop_reason 与累计 usage,不含 content
                    assembler.set_stop_reason(event.delta.stop_reason)
                    assembler.add_usage(event.usage)

                elif event.type == "message_stop":
                    # 消息终止信号(不含 content);此时内容块已累积完毕
                    message = assembler.build_message()

                    # 异步执行工具(StreamingToolExecutor)
                    tool_results = await self.execute_tools(
                        message.content,
                        app_state
                    )

                    # 更新应用状态
                    app_state.add_message(message)
                    app_state.add_tool_results(tool_results)

                    yield MessageEndEvent(message)

                    # 检查是否需要继续
                    if not tool_results:
                        app_state.mark_done()
                    # 否则继续下一轮(while 循环)

        # 返回最终结果
        yield AgentEndEvent(
            result=app_state.final_response,
            message_count=len(app_state.messages)
        )

class SessionRuntime:
    def run_session(self, session_id: str):
        """单个会话的运行时主循环"""
        while True:
            # 1. Intake:获取用户输入或等待事件
            user_message = self.intake_queue.get(timeout=30)
            if user_message is None:
                break  # 会话结束

            # 2. Context Assembly:组装上下文（每条新用户消息只组装一次）
            context = self.context_engine.assemble(
                session_id=session_id,
                user_input=user_message,
                system_prompt=self.system_prompt,
                tools_schema=self.get_tools_schema(),
                memory=self.memory_store.load(session_id)
            )

            # Initialize messages from context
            messages = context.messages.copy()

            # 3-4. 推理与工具调用循环：直到模型不再请求工具（4.1.3 的"工具调用耗尽"终止条件）
            while True:
                # 3. Inference:调用模型
                response = self.model_client.create_message(
                    messages=messages,
                    system=context.system_prompt,
                    tools=context.tools,
                    max_tokens=context.token_budget
                )

                # 4. Tool Execution Loop:处理本轮 response 中的所有工具调用
                tool_results = []
                for tool_use in response.content:
                    if isinstance(tool_use, ToolUseBlock):
                        result = self.execute_tool(
                            tool_use.name,
                            tool_use.input
                        )
                        tool_results.append(ToolResultBlock(
                            tool_use_id=tool_use.id,
                            content=result
                        ))

                # 没有工具调用 → 推理完成，跳出内层循环
                if not tool_results:
                    break

                # 有工具调用 → 把助手回复与工具结果包装成消息追加到 messages，继续内层推理
                # 注意：tool_result 块按 Anthropic Messages 约定属于 user 角色消息
                messages.append({"role": "assistant", "content": response.content})
                messages.append({"role": "user", "content": tool_results})

            # 5. Persistence:持久化（仅在内层循环结束、得到最终回复后执行一次）
            self.message_store.save(session_id, response)
            self.memory_store.maybe_consolidate(session_id)

            # 把最终响应交付给用户，回到 Intake 等待会话的下一条输入
            self.deliver(session_id, response)

class LoopTerminationChecker:
    def __init__(self, max_turns: int = 20, token_limit: int = 100000):
        self.max_turns = max_turns
        self.token_limit = token_limit

    def should_continue(self, state: AgentState) -> bool:
        """判断是否应该继续循环"""

        # 检查1:轮数上限
        if state.turn_count >= self.max_turns:
            state.termination_reason = "max_turns_reached"
            return False

        # 检查2:是否有工具调用
        last_message = state.messages[-1]
        if not self._has_tool_calls(last_message):
            state.termination_reason = "no_tool_calls"
            return False

        # 检查3:Token预算
        next_turn_tokens = self._estimate_tokens_for_next_turn(state)
        if next_turn_tokens > self.token_limit:
            state.termination_reason = "token_limit_exceeded"
            return False

        # 检查4:显式停止信号
        if state.stop_signal_received:
            state.termination_reason = "user_stop"
            return False

        return True

    def _has_tool_calls(self, message: Message) -> bool:
        """检查消息中是否有工具调用块"""
        for block in message.content:
            if isinstance(block, ToolUseBlock):
                return True
        return False

    def _estimate_tokens_for_next_turn(self, state: AgentState) -> int:
        """估计下一轮推理的Token使用"""
        # 简化版本,实际需要调用计数器
        history_tokens = len(state.messages) * 50  # 粗略估计
        new_input_tokens = 100  # 工具结果的平均大小
        return history_tokens + new_input_tokens

# 基础消息类型枚举
class MessageRole(Enum):
    USER = "user"           # 用户输入
    ASSISTANT = "assistant" # 智能体响应

class ContentBlockType(Enum):
    TEXT = "text"                     # 纯文本
    TOOL_USE = "tool_use"             # 工具调用请求
    TOOL_RESULT = "tool_result"       # 工具执行结果
    THINKING = "thinking"             # 推理过程(optional)
    IMAGE = "image"                   # 图像内容

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
import json

@dataclass
class TextBlock:
    """纯文本块"""
    type: str = "text"
    text: str = ""

    def to_dict(self) -> Dict[str, Any]:
        return {"type": self.type, "text": self.text}

@dataclass
class ToolUseBlock:
    """工具调用块:Agent 发起工具调用请求"""
    type: str = "tool_use"
    id: str = ""           # 工具使用的唯一ID,用于匹配结果
    name: str = ""         # 工具名称
    input: Dict[str, Any] = field(default_factory=dict)  # 工具参数,已解析为字典

    def __post_init__(self):
        if self.input is None:
            self.input = {}
        if not self.id:
            # 自动生成UUID
            import uuid
            self.id = f"tooluse_{uuid.uuid4().hex[:12]}"

    def to_dict(self) -> Dict[str, Any]:
        return {
            "type": self.type,
            "id": self.id,
            "name": self.name,
            "input": self.input
        }

@dataclass
class ToolResultBlock:
    """工具结果块:工具执行完成后的结果"""
    type: str = "tool_result"
    tool_use_id: str = ""  # 关联的 ToolUseBlock.id
    content: str = ""      # 工具执行结果(文本)
    is_error: bool = False # 是否为错误结果
    error_type: Optional[str] = None  # 错误类型(如果是错误)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "type": self.type,
            "tool_use_id": self.tool_use_id,
            "content": self.content,
            "is_error": self.is_error,
            "error_type": self.error_type
        }

@dataclass
class ThinkingBlock:
    """推理块:Agent 的内部思考过程(使用 extended thinking 时)"""
    type: str = "thinking"
    thinking: str = ""     # 思考内容

    def to_dict(self) -> Dict[str, Any]:
        return {"type": self.type, "thinking": self.thinking}

@dataclass
class Message:
    """完整的消息对象"""
    role: str  # "user" 或 "assistant"
    content: List[Any]  # 可以包含 TextBlock, ToolUseBlock, ToolResultBlock 等
    timestamp: datetime = None
    message_id: str = ""  # 唯一标识符

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now(timezone.utc)
        if not self.message_id:
            import uuid
            self.message_id = f"msg_{uuid.uuid4().hex[:12]}"

    @classmethod
    def user(cls, text: str) -> "Message":
        """工厂方法:创建用户消息"""
        return cls(
            role="user",
            content=[TextBlock(text=text)]
        )

    @classmethod
    def assistant(cls, content: List[Any]) -> "Message":
        """工厂方法:创建助手消息"""
        return cls(
            role="assistant",
            content=content
        )

    def has_tool_calls(self) -> bool:
        """检查是否包含工具调用"""
        return any(isinstance(block, ToolUseBlock) for block in self.content)

    def get_tool_calls(self) -> List[ToolUseBlock]:
        """提取所有工具调用块"""
        return [block for block in self.content if isinstance(block, ToolUseBlock)]

    def get_text(self) -> str:
        """提取所有文本块,拼接为单个字符串"""
        texts = [block.text for block in self.content if isinstance(block, TextBlock)]
        return "".join(texts)

    def to_dict(self) -> Dict[str, Any]:
        """序列化为字典"""
        return {
            "role": self.role,
            "content": [block.to_dict() for block in self.content],
            "timestamp": self.timestamp.isoformat(),
            "message_id": self.message_id
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Message":
        """从字典反序列化"""
        content = []
        for block_data in data.get("content", []):
            block_type = block_data.get("type")
            if block_type == "text":
                content.append(TextBlock(**block_data))
            elif block_type == "tool_use":
                content.append(ToolUseBlock(**block_data))
            elif block_type == "tool_result":
                content.append(ToolResultBlock(**block_data))
            elif block_type == "thinking":
                content.append(ThinkingBlock(**block_data))

        return cls(
            role=data["role"],
            content=content,
            timestamp=datetime.fromisoformat(data.get("timestamp", "")),
            message_id=data.get("message_id", "")
        )

from typing import Callable, List
from enum import Enum

class StateChangeEvent(Enum):
    MESSAGE_ADDED = "message_added"
    TOOL_EXECUTED = "tool_executed"
    STATE_RESET = "state_reset"

@dataclass
class AppState:
    """可变的应用状态"""
    session_id: str
    messages: List[Message] = field(default_factory=list)
    current_turn: int = 0
    token_count: int = 0
    _subscribers: List[Callable[[StateChangeEvent, Any], None]] = field(
        default_factory=list
    )

    def subscribe(self, callback: Callable[[StateChangeEvent, Any], None]):
        """订阅状态变化"""
        self._subscribers.append(callback)

    def _notify(self, event: StateChangeEvent, data: Any):
        """通知所有订阅者"""
        for callback in self._subscribers:
            try:
                callback(event, data)
            except Exception as e:
                print(f"Subscriber error: {e}")

    def add_message(self, message: Message):
        """添加消息(修改当前状态)"""
        self.messages.append(message)
        self.token_count += self._estimate_tokens(message)
        self.current_turn += 1
        # 通知订阅者
        self._notify(StateChangeEvent.MESSAGE_ADDED, message)

    def execute_tool(self, tool_name: str, result: str):
        """记录工具执行"""
        # ... 工具执行逻辑
        self._notify(StateChangeEvent.TOOL_EXECUTED, {
            "tool_name": tool_name,
            "result": result
        })

    def _estimate_tokens(self, message: Message) -> int:
        """粗略估计消息的Token数"""
        return int(len(message.get_text().split()) * 1.3)  # 简化估计

# 发布-订阅模式使用示例
app_state = AppState(session_id="sess_456")

def on_state_change(event: StateChangeEvent, data: Any):
    print(f"State changed: {event} -> {data}")

app_state.subscribe(on_state_change)
app_state.add_message(Message.user("Hello"))  # 触发 MESSAGE_ADDED 事件

from dataclasses import dataclass, field
from typing import List, Tuple

@dataclass(frozen=True)  # frozen=True 使对象不可变
class ImmutableSessionState:
    """不可变的会话状态"""
    session_id: str
    messages: Tuple[Message, ...] = field(default_factory=tuple)
    current_turn: int = 0
    token_count: int = 0
    last_updated: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def add_message(self, message: Message) -> "ImmutableSessionState":
        """添加消息,返回新的状态对象"""
        new_messages = tuple(list(self.messages) + [message])
        return ImmutableSessionState(
            session_id=self.session_id,
            messages=new_messages,
            current_turn=self.current_turn + 1,
            token_count=self.token_count + self._estimate_tokens(message),
            last_updated=datetime.now(timezone.utc)
        )

    def _estimate_tokens(self, message: Message) -> int:
        """粗略估计消息的Token数"""
        return len(message.get_text().split()) * 1.3  # 简化估计

# 不可变状态模式使用示例
state = ImmutableSessionState(session_id="sess_123")
print(f"初始状态 ID: {id(state)}")

state = state.add_message(Message.user("Hello"))
print(f"添加用户消息后的状态 ID: {id(state)}")  # 不同对象

state = state.add_message(Message.assistant([TextBlock(text="Hi there!")]))
print(f"添加助手消息后的状态 ID: {id(state)}")  # 又是不同对象

class MessageWindow:
    """消息窗口管理器"""

    def __init__(self, max_messages: int = 50, max_tokens: int = 100000):
        self.max_messages = max_messages
        self.max_tokens = max_tokens

    def filter_messages(self, messages: List[Message]) -> List[Message]:
        """根据约束条件过滤消息"""

        # 策略1:按消息数量限制
        if len(messages) > self.max_messages:
            # 保留最后 max_messages 条,丢弃早期消息
            messages = messages[-self.max_messages:]

        # 策略2:按Token数量限制
        total_tokens = sum(self._estimate_tokens(m) for m in messages)
        if total_tokens > self.max_tokens:
            # 移除早期消息直到满足Token限制
            while messages and total_tokens > self.max_tokens:
                removed = messages.pop(0)
                total_tokens -= self._estimate_tokens(removed)

        return messages

    def _estimate_tokens(self, message: Message) -> int:
        """估计消息的Token数"""
        text = message.get_text()
        return len(text.split()) * 1.3

    def insert_system_context(self, messages: List[Message],
                             system_blocks: List[TextBlock]) -> List[Message]:
        """在消息历史前插入系统上下文"""
        result = []
        # 添加系统消息(不计入限制)
        for block in system_blocks:
            result.append(Message.assistant([block]))
        # 添加过滤后的消息历史
        result.extend(self.filter_messages(messages))
        return result

class StreamingToolExecutor:
    """流式工具执行器:在响应流中处理工具调用"""

    def __init__(self, tool_registry: ToolRegistry, max_concurrent: int = 5):
        self.tool_registry = tool_registry
        self.max_concurrent = max_concurrent
        self._pending_executions: Dict[str, asyncio.Task] = {}

    async def execute_tools(
        self,
        tool_uses: List[ToolUseBlock],
        app_state: AppState
    ) -> Dict[str, ToolResultBlock]:
        """
        并发执行多个工具调用,返回所有结果
        使用 asyncio.Semaphore 限制并发数
        """
        semaphore = asyncio.Semaphore(self.max_concurrent)
        tasks = []

        for tool_use in tool_uses:
            task = asyncio.create_task(
                self._execute_single_tool(tool_use, app_state, semaphore)
            )
            tasks.append(task)
            self._pending_executions[tool_use.id] = task

        # 并发等待所有工具完成
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 组织结果
        tool_results = {}
        for tool_use, result in zip(tool_uses, results):
            if isinstance(result, Exception):
                tool_results[tool_use.id] = ToolResultBlock(
                    tool_use_id=tool_use.id,
                    content=f"Tool execution error: {str(result)}",
                    is_error=True,
                    error_type=type(result).__name__
                )
            else:
                tool_results[tool_use.id] = result

        return tool_results

    async def _execute_single_tool(
        self,
        tool_use: ToolUseBlock,
        app_state: AppState,
        semaphore: asyncio.Semaphore
    ) -> ToolResultBlock:
        """执行单个工具"""
        async with semaphore:
            try:
                # 1. 查找工具
                tool = self.tool_registry.get(tool_use.name)
                if not tool:
                    raise ToolNotFoundError(f"Tool '{tool_use.name}' not found")

                # 2. 权限检查
                if not tool.check_permissions(app_state):
                    raise PermissionDeniedError(
                        f"Permission denied for tool '{tool_use.name}'"
                    )

                # 3. 执行工具
                result = await tool.call(tool_use.input)

                # 4. 组织结果
                return ToolResultBlock(
                    tool_use_id=tool_use.id,
                    content=str(result),
                    is_error=False
                )

            except Exception as e:
                return ToolResultBlock(
                    tool_use_id=tool_use.id,
                    content=str(e),
                    is_error=True,
                    error_type=type(e).__name__
                )

    def get_tool_progress(self, tool_use_id: str) -> Optional[ToolProgress]:
        """获取某个工具的执行进度(如果可用)"""
        task = self._pending_executions.get(tool_use_id)
        if not task:
            return None
        # 实现细节:从工具对象的进度属性中提取(如果工具支持)
        # 通常通过工具返回的结果对象或单独的进度查询接口获取
        # 这是一个简化的实现,实际需要根据具体工具的API调整
        if hasattr(task, '_progress'):
            return task._progress
        return None

class StreamingResponseHandler:
    """处理 API 流式响应"""

    def handle_stream(self, stream):
        """逐个处理流中的事件"""
        accumulated_input_json = ""  # 累积工具参数的 JSON
        current_tool_name = None  # content_block_stop 事件只携带 index，工具名须在 start 时记录

        for event in stream:
            if event.type == "content_block_start":
                if event.content_block.type == "text":
                    # 开始新的文本块
                    pass

                elif event.content_block.type == "tool_use":
                    # 工具调用开始：记下工具名，供 stop 事件使用
                    current_tool_name = event.content_block.name
                    accumulated_input_json = ""

            elif event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    # 文本增量,直接转发给前端
                    self.send_to_frontend(TextDeltaEvent(event.delta.text))

                elif event.delta.type == "input_json_delta":
                    # 累积 JSON 参数
                    accumulated_input_json += event.delta.partial_json

            elif event.type == "content_block_stop":
                if current_tool_name is not None:
                    # 工具调用完成,此时有完整的 accumulated_input_json
                    tool_input = json.loads(accumulated_input_json)

                    # 输入完整后执行工具(可能阻塞,但用户已看到前面的文本增量)
                    result = self.execute_tool(current_tool_name, tool_input)
                    self.send_to_frontend(ToolResultEvent(result))
                    current_tool_name = None

import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, AsyncIterator, Dict
from enum import Enum

class EventType(Enum):
    AGENT_START = "agent_start"
    TURN_START = "turn_start"
    CONTENT_BLOCK_START = "content_block_start"
    TEXT_DELTA = "text_delta"
    TOOL_USE_START = "tool_use_start"
    TOOL_INPUT_DELTA = "tool_input_delta"
    TOOL_USE_END = "tool_use_end"
    TOOL_RESULT = "tool_result"
    CONTENT_BLOCK_END = "content_block_end"
    MESSAGE_END = "message_end"
    TURN_END = "turn_end"
    AGENT_END = "agent_end"
    ERROR = "error"

@dataclass
class Event:
    """事件基类"""
    event_type: EventType
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    metadata: Dict[str, Any] = field(default_factory=dict)

    def to_json(self) -> str:
        """序列化为 JSON(用于流式传输)"""
        return json.dumps({
            "event_type": self.event_type.value,
            "timestamp": self.timestamp.isoformat(),
            "metadata": self.metadata
        })

class QueryEngine:
    """生成事件流的查询引擎"""

    async def submit_message(self, user_input: str) -> AsyncIterator[Event]:
        """异步生成器:逐个 yield 事件"""
        try:
            yield Event(EventType.AGENT_START, metadata={
                "user_input_length": len(user_input)
            })

            app_state = AppState(session_id=str(uuid.uuid4()))
            turn_number = 0

            while True:
                turn_number += 1
                yield Event(EventType.TURN_START, metadata={
                    "turn_number": turn_number,
                    "context_tokens": app_state.token_count
                })

                # 构建上下文
                messages = await self._build_messages(app_state)
                system_prompt = self._get_system_prompt()

                # 流式调用模型
                tool_uses = []
                async for event in self._stream_model_response(
                    messages=messages,
                    system=system_prompt
                ):
                    # 转发模型流中的文本和工具使用事件
                    yield event

                    if event.event_type == EventType.TOOL_USE_END:
                        tool_uses.append(event.metadata["tool_use"])

                # 工具输入完整后执行工具；多个工具可以并发，但需要等待结果进入下一轮
                if tool_uses:
                    tool_results = await self.executor.execute_tools(
                        tool_uses, app_state
                    )

                    # 发出工具结果事件
                    for tool_use_id, result in tool_results.items():
                        yield Event(EventType.TOOL_RESULT, metadata={
                            "tool_use_id": tool_use_id,
                            "is_error": result.is_error,
                            "content_length": len(result.content)
                        })

                    # 更新应用状态以继续下一轮推理
                    app_state.add_tool_results(tool_results)

                else:
                    # 没有工具调用,循环结束
                    break

                yield Event(EventType.TURN_END, metadata={
                    "turn_number": turn_number,
                    "has_tool_calls": bool(tool_uses)
                })

            yield Event(EventType.AGENT_END, metadata={
                "final_response": app_state.get_final_response(),
                "turn_count": turn_number,
                "message_count": len(app_state.messages)
            })

        except Exception as e:
            yield Event(EventType.ERROR, metadata={
                "error_type": type(e).__name__,
                "error_message": str(e)
            })

class BackpressureQueue:
    """带背压的事件队列"""

    def __init__(self, max_queue_size: int = 100):
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue_size)

    async def put_event(self, event: Event):
        """发送事件,如果队列满则等待"""
        try:
            self.queue.put_nowait(event)
        except asyncio.QueueFull:
            # 队列满,等待消费者处理
            await self.queue.put(event)

    async def get_event(self) -> Event:
        """接收事件,带超时"""
        return await asyncio.wait_for(
            self.queue.get(),
            timeout=30.0  # 30秒超时
        )

    def queue_size(self) -> int:
        """获取当前队列大小"""
        return self.queue.qsize()

import asyncio

class ToolExecutionError(Exception):
    """工具执行错误"""
    def __init__(self, tool_name: str, message: str,
                 error_type: str = None, retry_count: int = 0):
        self.tool_name = tool_name
        self.message = message
        self.error_type = error_type or type(self).__name__
        self.retry_count = retry_count
        super().__init__(message)

class ToolTimeoutError(ToolExecutionError):
    """工具执行超时"""
    pass

class ToolPermissionError(ToolExecutionError):
    """工具权限不足"""
    pass

# 处理示例
async def execute_tool_with_recovery(
    tool_use: ToolUseBlock,
    executor: ToolExecutor,
    max_retries: int = 3
) -> ToolResultBlock:
    """执行工具,带重试机制"""

    for attempt in range(max_retries):
        try:
            result = await executor.execute(tool_use.name, tool_use.input)
            return ToolResultBlock(
                tool_use_id=tool_use.id,
                content=str(result),
                is_error=False
            )

        except ToolTimeoutError as e:
            # 超时:使用指数退避重试
            if attempt < max_retries - 1:
                backoff_seconds = 2 ** attempt  # 1, 2, 4 秒
                await asyncio.sleep(backoff_seconds)
                continue
            else:
                # 最后一次重试失败,返回错误
                return ToolResultBlock(
                    tool_use_id=tool_use.id,
                    content=f"Tool timeout after {max_retries} retries: {str(e)}",
                    is_error=True,
                    error_type="ToolTimeoutError"
                )

        except ToolPermissionError as e:
            # 权限错误:不重试,立即反馈
            return ToolResultBlock(
                tool_use_id=tool_use.id,
                content=f"Permission denied: {str(e)}",
                is_error=True,
                error_type="ToolPermissionError"
            )

        except ToolExecutionError as e:
            # 其他工具错误:重试一次
            if attempt < max_retries - 1:
                continue
            else:
                return ToolResultBlock(
                    tool_use_id=tool_use.id,
                    content=f"Tool execution failed: {str(e)}",
                    is_error=True,
                    error_type=e.error_type
                )

        except Exception as e:
            # 未预期的异常
            return ToolResultBlock(
                tool_use_id=tool_use.id,
                content=f"Unexpected error: {str(e)}",
                is_error=True,
                error_type=type(e).__name__
            )

class ModelAPIError(Exception):
    """模型 API 错误"""
    pass

class RateLimitError(ModelAPIError):
    """速率限制"""
    def __init__(self, retry_after: int = None):
        self.retry_after = retry_after or 60
        super().__init__(f"Rate limited, retry after {self.retry_after}s")

class ModelTimeoutError(ModelAPIError):
    """模型调用超时"""
    pass

class RetryPolicy:
    """重试策略"""

    def __init__(self,
                 max_retries: int = 3,
                 initial_backoff: int = 1,
                 max_backoff: int = 60,
                 exponential_base: float = 2.0):
        self.max_retries = max_retries
        self.initial_backoff = initial_backoff
        self.max_backoff = max_backoff
        self.exponential_base = exponential_base

    async def execute_with_retry(self,
                                 coro_fn,
                                 *args, **kwargs) -> Any:
        """执行异步操作,带重试"""

        for attempt in range(self.max_retries):
            try:
                return await coro_fn(*args, **kwargs)

            except RateLimitError as e:
                # 速率限制:等待指定时间
                if attempt < self.max_retries - 1:
                    await asyncio.sleep(e.retry_after)
                    continue
                else:
                    raise

            except ModelTimeoutError as e:
                # 超时:指数退避
                if attempt < self.max_retries - 1:
                    backoff = min(
                        self.initial_backoff * (self.exponential_base ** attempt),
                        self.max_backoff
                    )
                    await asyncio.sleep(backoff)
                    continue
                else:
                    raise

            except ModelAPIError as e:
                # 其他 API 错误:重试
                if attempt < self.max_retries - 1:
                    backoff = min(
                        self.initial_backoff * (self.exponential_base ** attempt),
                        self.max_backoff
                    )
                    await asyncio.sleep(backoff)
                    continue
                else:
                    raise

# API重试策略使用示例
retry_policy = RetryPolicy()

async def call_model_with_retry(engine: QueryEngine, message: str):
    try:
        result = await retry_policy.execute_with_retry(
            engine.infer,
            messages=[Message.user(message)]
        )
        return result
    except ModelAPIError as e:
        # 重试失败,返回降级响应
        return Message.assistant([TextBlock(
            text=f"Sorry, I encountered an error: {str(e)}"
        )])

class OutputParsingError(Exception):
    """输出解析错误"""
    pass

class ToolParameterValidationError(OutputParsingError):
    """工具参数验证失败"""
    def __init__(self, tool_name: str, errors: List[str]):
        self.tool_name = tool_name
        self.errors = errors
        super().__init__(
            f"Tool '{tool_name}' parameter validation failed: {errors}"
        )

def validate_tool_parameters(
    tool_name: str,
    parameters: Dict[str, Any],
    schema: Dict[str, Any]
) -> Tuple[bool, List[str]]:
    """验证工具参数是否符合 Schema"""

    errors = []

    # 检查必需参数
    required = schema.get("required", [])
    for param_name in required:
        if param_name not in parameters:
            errors.append(f"Missing required parameter: {param_name}")

    # 检查参数类型
    properties = schema.get("properties", {})
    for param_name, param_value in parameters.items():
        if param_name not in properties:
            errors.append(f"Unknown parameter: {param_name}")
            continue

        param_schema = properties[param_name]
        expected_type = param_schema.get("type")

        if expected_type == "string" and not isinstance(param_value, str):
            errors.append(
                f"Parameter '{param_name}' should be string, got {type(param_value).__name__}"
            )
        elif expected_type == "integer" and not isinstance(param_value, int):
            errors.append(
                f"Parameter '{param_name}' should be integer, got {type(param_value).__name__}"
            )
        elif expected_type == "array" and not isinstance(param_value, list):
            errors.append(
                f"Parameter '{param_name}' should be array, got {type(param_value).__name__}"
            )

    return len(errors) == 0, errors

# 在工具执行前进行验证
async def execute_tool_validated(tool_use: ToolUseBlock,
                                 tool_registry: ToolRegistry) -> ToolResultBlock:
    """执行工具前进行参数验证"""

    tool = tool_registry.get(tool_use.name)
    if not tool:
        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=f"Tool '{tool_use.name}' not found",
            is_error=True,
            error_type="ToolNotFoundError"
        )

    # 验证参数
    schema = tool.get_input_schema()
    is_valid, errors = validate_tool_parameters(
        tool_use.name,
        tool_use.input,
        schema
    )

    if not is_valid:
        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=f"Parameter validation failed:\n" + "\n".join(errors),
            is_error=True,
            error_type="ParameterValidationError"
        )

    # 参数有效,执行工具
    try:
        result = await tool.call(tool_use.input)
        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=str(result),
            is_error=False
        )
    except Exception as e:
        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=str(e),
            is_error=True,
            error_type=type(e).__name__
        )

class ContextOverflowError(Exception):
    """上下文溢出"""
    pass

class TokenBudgetExceededError(ContextOverflowError):
    """Token预算超支"""
    def __init__(self, current_tokens: int, budget: int):
        self.current_tokens = current_tokens
        self.budget = budget
        super().__init__(
            f"Token budget exceeded: {current_tokens} > {budget}"
        )

class ContextCompressor:
    """上下文压缩器"""

    def __init__(self, max_summary_length: int = 500):
        self.max_summary_length = max_summary_length

    def compress_message_history(
        self,
        messages: List[Message],
        target_tokens: int,
        summarizer = None  # 可选的摘要函数
    ) -> List[Message]:
        """压缩消息历史以满足Token预算"""

        total_tokens = sum(self._estimate_tokens(m) for m in messages)

        if total_tokens <= target_tokens:
            return messages  # 无需压缩

        # 策略1:先对过长的 Assistant 响应做摘要（信息损失最小）
        if summarizer:
            summarized_messages = []
            for message in messages:
                if message.role == "assistant":
                    # 摘要 Assistant 的长响应
                    original_text = message.get_text()
                    if len(original_text) > self.max_summary_length:
                        summary = summarizer(original_text,
                                           max_length=self.max_summary_length)
                        message = Message.assistant([TextBlock(text=summary)])
                summarized_messages.append(message)

            messages = summarized_messages
            total_tokens = sum(self._estimate_tokens(m) for m in messages)

        # 策略2:摘要后仍超出预算,再从最早的消息开始移除
        while messages and total_tokens > target_tokens:
            removed = messages.pop(0)
            total_tokens -= self._estimate_tokens(removed)

        return messages

    def _estimate_tokens(self, message: Message) -> int:
        """估计消息的Token数"""
        text = message.get_text()
        return int(len(text.split()) * 1.3)  # 粗略估计

# 上下文压缩使用示例
compressor = ContextCompressor()

async def infer_with_context_management(
    engine: QueryEngine,
    messages: List[Message],
    token_budget: int = 100000
) -> Message:
    """推理,带自动的上下文管理"""

    estimated_tokens = compressor._estimate_tokens(
        Message.assistant([TextBlock(text="".join(m.get_text() for m in messages))])
    )

    if estimated_tokens > token_budget * 0.8:  # 80% 阈值
        messages = compressor.compress_message_history(
            messages,
            target_tokens=int(token_budget * 0.5)
        )

    try:
        return await engine.infer(messages)
    except TokenBudgetExceededError as e:
        # Token预算溢出,再次压缩
        messages = compressor.compress_message_history(
            messages,
            target_tokens=int(token_budget * 0.3)
        )
        return await engine.infer(messages)

class ErrorAsObservationHandler:
    """将错误作为观察反馈给 Agent"""

    def handle_tool_error(self, tool_use: ToolUseBlock, error: Exception):
        """处理工具错误"""
        # 而不是抛出异常,构造一个 ToolResultBlock,表示错误
        error_message = f"[Tool Error]\nTool: {tool_use.name}\n"
        error_message += f"Error Type: {type(error).__name__}\n"
        error_message += f"Error Message: {str(error)}\n"
        error_message += "Please analyze the error and try a different approach."

        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=error_message,
            is_error=True,
            error_type=type(error).__name__
        )

    def handle_api_error(self, error: ModelAPIError) -> Message:
        """处理 API 错误"""
        # 返回一条特殊的 Assistant 消息,告知智能体发生了错误
        return Message.assistant([TextBlock(text=(
            f"An error occurred while processing your request:\n"
            f"Error: {str(error)}\n"
            f"The system will retry the request automatically."
        ))])

    def handle_context_overflow(self,
                               messages: List[Message],
                               token_budget: int) -> List[Message]:
        """处理上下文溢出"""
        # 通过压缩历史来恢复
        return self.compress_history(messages, token_budget)

class CircuitBreakerState(Enum):
    CLOSED = "closed"      # 正常,允许请求
    OPEN = "open"         # 失败过多,拒绝请求
    HALF_OPEN = "half_open"  # 尝试恢复,允许一个请求

class CircuitBreaker:
    """熔断器模式"""

    def __init__(self, failure_threshold: int = 5,
                 recovery_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.state = CircuitBreakerState.CLOSED
        self.last_failure_time = None

    async def call(self, coro_fn, *args, **kwargs):
        """通过熔断器调用函数"""

        # 检查是否应该打开熔断器
        if self.state == CircuitBreakerState.OPEN:
            if self._should_attempt_recovery():
                self.state = CircuitBreakerState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN, rejecting request")

        try:
            result = await coro_fn(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        """成功后的处理"""
        self.failure_count = 0
        self.state = CircuitBreakerState.CLOSED

    def _on_failure(self):
        """失败后的处理"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitBreakerState.OPEN

    def _should_attempt_recovery(self) -> bool:
        """判断是否应该尝试恢复"""
        if not self.last_failure_time:
            return False
        return time.time() - self.last_failure_time > self.recovery_timeout