# 传统串行模式
response = await client.messages.create(
    model="claude-sonnet-4-20250514",
    tools=tools,
    messages=[{"role": "user", "content": "查询北京天气并发邮件"}]
)
# 必须等response完整返回后才开始执行
tool_result = await execute_tool(response.tool_calls[0])

# Eager Calling模式（伪代码）
async def eager_tool_call(stream):
    partial_params = {}
    async for event in stream:
        if event.type == "tool_use_partial":
            partial_params.update(event.partial_json)
            # 参数达到可执行阈值时立即启动
            if is_executable(partial_params):
                # 启动工具执行，不等参数全部到达
                task = asyncio.create_task(
                    execute_with_defaults(partial_params)
                )
        if event.type == "tool_use_complete":
            # 参数完成，补全缺失字段
            task.finalize(event.full_params)
            return await task

# PASTE模式库结构
pattern_bank = {
    "flight_booking": {
        "sequence": ["search_flights", "get_price", "book_flight"],
        "trigger_phrases": ["订机票", "book flight", "航班"],
        "param_transfer": {
            # 第1步输出→第2步输入的映射
            "search_flights.flight_id": "get_price.flight_id",
            "get_price.total": "book_flight.amount"
        },
        "confidence": 0.87
    }
}

# 推测执行逻辑
async def paste_execute(user_message, model_stream):
    # 1. 识别可能的模式
    predicted = pattern_bank.match(user_message)
    if predicted.confidence > THRESHOLD:
        # 2. 预执行下一步（与模型推理并行）
        pre_result = await speculative_execute(
            predicted.next_tool,
            estimated_params=predicted.estimate_from_history()
        )
    # 3. 模型实际输出后验证
    actual_call = await parse_model_output(model_stream)
    if matches_prediction(actual_call, predicted):
        return pre_result  # 命中，直接使用
    else:
        cancel(pre_result)  # 未命中，回退到正常执行

# Speculative Actions 核心循环
class SpeculativeActionEngine:
    def __init__(self, predictor, verifier):
        self.predictor = predictor  # 基于历史预测下一动作
        self.verifier = verifier    # 验证推测是否正确

    async def step(self, state, actual_action_stream):
        # 路径A：推测执行（立即启动）
        predicted_action = self.predictor.predict_next(state)
        speculative_task = asyncio.create_task(
            self.execute_speculative(predicted_action)
        )

        # 路径B：等待模型实际输出
        actual_action = await actual_action_stream

        # 验证：推测是否与实际一致？
        if self.verifier.matches(predicted_action, actual_action):
            result = await speculative_task  # 命中！零额外等待
            return result
        else:
            speculative_task.cancel()  # 未命中，取消推测
            return await self.execute_actual(actual_action)

import anthropic

client = anthropic.Anthropic()

# 启用细粒度工具流式
with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=[{
        "name": "get_weather",
        "description": "获取天气信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "date": {"type": "string"}
            }
        }
    }],
    messages=[{"role": "user", "content": "北京明天天气"}]
) as stream:
    for event in stream:
        if event.type == "tool_use_partial":
            # 参数片段到达，可提前准备
            if "city" in event.partial_json:
                print(f"城市已确定: {event.partial_json['city']}")
                # 预热天气API连接
                prewarm_weather_api(event.partial_json['city'])

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="gpt-4o",
    tools=[
        {"type": "function", "function": {
            "name": "get_weather", "parameters": {...}
        }},
        {"type": "function", "function": {
            "name": "get_stock_price", "parameters": {...}
        }},
        {"type": "function", "function": {
            "name": "get_news", "parameters": {...}
        }}
    ],
    input="今天天气如何？苹果股价多少？最新新闻有哪些？"
)

# response中包含3个并行的tool_calls
# 三个工具同时执行，而非串行等待
import asyncio
results = await asyncio.gather(*[
    execute_tool(call) for call in response.output
])

from speculative_tools import SpeculativeExecutor

executor = SpeculativeExecutor(
    model_client=your_llm_client,
    tools=your_tools,
    strategy="trajectory_aware",  # 基于轨迹感知的预测
    max_speculative_depth=2,       # 最多推测2步
    cancellation_timeout=500       # 500ms内未确认则取消
)

result = await executor.run(
    messages=[{"role": "user", "content": "帮我分析这份报告并生成摘要"}]
)
# 引擎内部流程：
# 1. 模型输出 "read_file" → 立即推测下一步是 "summarize"
# 2. 预启动 summarize 的准备工作
# 3. 模型实际输出 "summarize" → 命中！使用预准备结果

graph TB
    subgraph "传统串行模式"
        A1[模型推理] -->|等待完成| A2[参数解析]
        A2 -->|等待完成| A3[工具执行]
        A3 -->|等待完成| A4[结果处理]
        A4 -->|等待完成| A5[下一轮推理]
    end

    subgraph "重叠执行模式"
        B1[模型推理] --> B2[参数流式输出]
        B2 --> B3[FGTS: 参数增量传输]
        B3 --> B4[Eager: 参数就绪即执行]
        B4 --> B5[工具执行]
        B5 --> B6[结果预解析]
        B2 -.->|PASTE/SA: 并行推测| B5
        B1 -.->|模式预测| SPEC[推测执行]
        SPEC -.->|命中则跳过等待| B6
        B5 -.->|并行调用| PAR[并行工具2...N]
        B6 --> B7[下一轮推理]
    end

    style B1 fill:#4A90D9,color:#fff
    style B4 fill:#E8833A,color:#fff
    style SPEC fill:#7B68EE,color:#fff
    style PAR fill:#2ECC71,color:#fff