上下文窗口与 Function Calling：AI 的记忆与双手

{
  "model": "gpt-4o",
  "messages": [
    { "role": "user", "content": "上海今天天气怎么样？需要带伞吗？" }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "查询指定城市当前的实时天气情况",
        "parameters": {
          "type": "object",
          "properties": {
            "city": {
              "type": "string",
              "description": "城市名称，例如：上海"
            },
            "unit": {
              "type": "string",
              "enum": ["celsius", "fahrenheit"],
              "description": "返回的温度单位"
            }
          },
          "required": ["city"]
        }
      }
    }
  ]
}

{
  "choices": [
    {
      "message": {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_9f2k7a",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": {
                "city": "上海",
                "unit": "celsius"
              }
            }
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ]
}

{
  "role": "tool",
  "tool_call_id": "call_9f2k7a",
  "content": { "city": "上海", "temperature": 26, "condition": "小雨" }
}

每token显存 = 2 × 层数 × KV注意力头数 × 每头维度 × 精度字节数

其中：
- 系数 2 代表 Key 和 Value 各存一份
- 精度字节数：FP16/BF16 为 2，FP8 为 1，INT8 为 1

每token显存 = 2 × 40 × 40 × 128 × 2 字节
           = 1,638,400 字节
           ≈ 1.56 MB / token

from transformers import AutoModelForCausalLM

# 通过 rope_scaling 把原生窗口按 YaRN 方式扩展
model = AutoModelForCausalLM.from_pretrained(
    "your-org/your-7b-model",
    rope_scaling={
        "type": "yarn",
        "factor": 4.0,               # 扩展倍数：4K -> 16K
        "original_max_position_embeddings": 4096,
    },
    torch_dtype="bfloat16",
    device_map="auto",
)

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一名资深法律顾问，以下是需要遵循的完整合同审查规范……",
        },
        {
            "type": "text",
            "text": LONG_CONTRACT_TEMPLATE,  # 几万 token 的长文档
            "cache_control": {"type": "ephemeral"},  # 在此处打缓存断点
        },
    ],
    messages=[
        {"role": "user", "content": "请审查这份新合同的第3条是否合规。"}
    ],
)

# 首次请求写入缓存（略贵）；5分钟内的后续请求命中缓存，
# 前缀部分按约 1/10 计费，TTFT 也明显下降。
usage = response.usage
print("缓存写入 token:", usage.cache_creation_input_tokens)
print("缓存命中 token:", usage.cache_read_input_tokens)
print("常规输入 token:", usage.input_tokens)

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,           # 每块约800字符
    chunk_overlap=120,        # 相邻块重叠120字符，避免边界语义割裂
    separators=["\n\n", "\n", "。", "！", "？", " ", ""],  # 逐级回退
    length_function=len,
)

docs = splitter.split_text(long_document)
print(f"共切分出 {len(docs)} 块")
for i, d in enumerate(docs[:3]):
    print(f"--- 第{i}块（{len(d)}字）---")
    print(d)

from sentence_transformers import CrossEncoder

# 1) 向量检索粗召回（伪代码，candidates 来自向量库 top-30）
candidates = vector_store.similarity_search(query, k=30)

# 2) 交叉编码器精排
reranker = CrossEncoder("BAAI/bge-reranker-large")
pairs = [(query, doc.page_content) for doc in candidates]
scores = reranker.predict(pairs)

# 3) 按精排分数取前 5 条注入上下文
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
top_docs = [doc for doc, score in ranked[:5]]

context = "\n\n".join(d.page_content for d in top_docs)
prompt = f"根据以下资料回答问题：\n{context}\n\n问题：{query}"

from dataclasses import dataclass, field
from typing import List, Dict
import tiktoken


@dataclass
class Message:
    role: str
    content: str


class HybridMemory:
    """混合记忆：近期原文滑窗 + 中期摘要 + 长期向量检索。"""

    def __init__(
        self,
        model: str = "gpt-4o",
        max_window_tokens: int = 4000,   # 滑窗预算
        summary_trigger: int = 3000,     # 超过该阈值触发摘要
        recent_keep: int = 6,            # 摘要后保留的最近消息条数
    ):
        self.enc = tiktoken.encoding_for_model(model)
        self.max_window_tokens = max_window_tokens
        self.summary_trigger = summary_trigger
        self.recent_keep = recent_keep
        self.messages: List[Message] = []   # 近期原文
        self.summary: str = ""              # 中期摘要
        self.vector_store = SimpleVectorStore()  # 长期记忆（见下）

    def _count(self, text: str) -> int:
        return len(self.enc.encode(text))

    def _window_tokens(self) -> int:
        return sum(self._count(m.content) for m in self.messages)

    def add(self, role: str, content: str):
        self.messages.append(Message(role, content))
        # 把每条消息也写入长期向量记忆，便于日后语义检索
        self.vector_store.add(content, metadata={"role": role})
        if self._window_tokens() > self.summary_trigger:
            self._compress()

    def _compress(self):
        """把较早的消息摘要化，只保留最近 recent_keep 条原文。"""
        to_summarize = self.messages[:-self.recent_keep]
        if not to_summarize:
            return
        joined = "\n".join(f"{m.role}: {m.content}" for m in to_summarize)
        new_summary = llm_summarize(
            f"已有摘要：\n{self.summary}\n\n"
            f"请把下列新增对话合并进摘要，保留关键事实、决定、用户偏好，"
            f"控制在300字以内：\n{joined}"
        )
        self.summary = new_summary
        self.messages = self.messages[-self.recent_keep:]

    def build_context(self, user_query: str) -> List[Dict]:
        """组装最终发给模型的消息列表。"""
        ctx: List[Dict] = []
        # 1) 长期记忆：按当前问题语义检索最相关的历史片段
        recalled = self.vector_store.search(user_query, k=3)
        if recalled:
            ctx.append({
                "role": "system",
                "content": "以下是可能相关的历史记忆：\n" + "\n".join(recalled),
            })
        # 2) 中期记忆：摘要
        if self.summary:
            ctx.append({"role": "system", "content": f"对话摘要：{self.summary}"})
        # 3) 近期记忆：原文滑窗
        for m in self.messages:
            ctx.append({"role": m.role, "content": m.content})
        # 4) 当前问题
        ctx.append({"role": "user", "content": user_query})
        return ctx

import numpy as np


class SimpleVectorStore:
    def __init__(self):
        self.texts: List[str] = []
        self.vectors: List[np.ndarray] = []

    def add(self, text: str, metadata: dict = None):
        vec = embed(text)  # 调用嵌入模型，返回归一化向量
        self.texts.append(text)
        self.vectors.append(vec)

    def search(self, query: str, k: int = 3) -> List[str]:
        if not self.vectors:
            return []
        q = embed(query)
        sims = [float(np.dot(q, v)) for v in self.vectors]
        top_idx = np.argsort(sims)[::-1][:k]
        return [self.texts[i] for i in top_idx if sims[i] > 0.7]  # 相似度阈值过滤

import json
from openai import OpenAI

client = OpenAI()

# 1) 定义工具的 JSON Schema
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的实时天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名"},
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_flights",
            "description": "查询两地之间某天的航班",
            "parameters": {
                "type": "object",
                "properties": {
                    "origin": {"type": "string"},
                    "destination": {"type": "string"},
                    "date": {"type": "string", "description": "YYYY-MM-DD"},
                },
                "required": ["origin", "destination", "date"],
            },
        },
    },
]

# 2) 工具的真实实现（宿主侧）
def get_weather(city: str) -> dict:
    return {"city": city, "temp": 26, "condition": "小雨"}

def search_flights(origin: str, destination: str, date: str) -> dict:
    return {"flights": [{"no": "MU5101", "price": 880, "depart": "08:30"}]}

TOOL_IMPL = {"get_weather": get_weather, "search_flights": search_flights}


def run_agent(user_query: str, max_steps: int = 6) -> str:
    messages = [
        {"role": "system", "content": "你是一个出行助手，善用工具获取实时信息。"},
        {"role": "user", "content": user_query},
    ]
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",        # 让模型自主决定是否调用工具
            parallel_tool_calls=True,  # 允许一次返回多个并行调用
        )
        msg = resp.choices[0].message
        messages.append(msg)

        # 模型没有再要工具，说明它给出了最终答案
        if not msg.tool_calls:
            return msg.content

        # 逐个执行模型请求的工具（可能是多个并行调用）
        for call in msg.tool_calls:
            name = call.function.name
            args = json.loads(call.function.arguments)  # 注意：arguments 是字符串
            try:
                result = TOOL_IMPL[name](**args)
            except Exception as e:
                result = {"error": str(e)}
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })
    return "已达到最大步数上限，仍未得到最终答案。"


print(run_agent("我明天从上海去北京，先告诉我两地天气，再帮我看看航班。"))

REACT_SYSTEM = """你要用如下循环解决问题：
Thought: 分析现状，思考下一步该做什么
Action: 选择一个工具并给出参数
Observation: （由系统填入工具返回结果）
... 循环若干轮 ...
Thought: 我已经掌握足够信息
Final Answer: 给出最终答案

可用工具：
- search(query): 联网搜索
- calculator(expr): 计算数学表达式
"""

def react_loop(question: str, max_iters: int = 8) -> str:
    scratchpad = ""
    for _ in range(max_iters):
        prompt = f"{REACT_SYSTEM}\n\n问题：{question}\n{scratchpad}Thought:"
        text = llm_complete(prompt, stop=["Observation:"])
        scratchpad += "Thought:" + text

        if "Final Answer:" in text:
            return text.split("Final Answer:")[-1].strip()

        # 解析 Action 行，执行工具
        action, arg = parse_action(text)       # 从文本中抽取工具名与参数
        observation = TOOLS[action](arg)
        scratchpad += f"\nObservation: {observation}\n"
    return "超出最大迭代次数。"

from pydantic import BaseModel, Field, field_validator
from typing import Literal
from openai import OpenAI

client = OpenAI()


class Invoice(BaseModel):
    vendor: str = Field(description="开票方公司名称")
    amount: float = Field(description="金额，单位元", gt=0)
    currency: Literal["CNY", "USD", "EUR"] = "CNY"
    invoice_date: str = Field(description="开票日期 YYYY-MM-DD")
    items: list[str] = Field(description="商品或服务明细")

    @field_validator("invoice_date")
    @classmethod
    def check_date(cls, v: str) -> str:
        import re
        if not re.match(r"^\d{4}-\d{2}-\d{2}$", v):
            raise ValueError("日期格式必须为 YYYY-MM-DD")
        return v


# 用 SDK 的结构化输出能力，直接把 Pydantic 模型作为响应格式
completion = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是发票信息抽取助手。"},
        {"role": "user", "content": "从这段文字抽取发票信息：上海某某科技有限公司，"
                                     "2026年8月3日开具，金额12800元，内容为云服务费。"},
    ],
    response_format=Invoice,   # 直接传 Pydantic 模型
)

invoice: Invoice = completion.choices[0].message.parsed
print(invoice.vendor, invoice.amount, invoice.invoice_date)
# 拿到的 invoice 已经是通过 Pydantic 校验的强类型对象

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-tools")


@mcp.tool()
def query_order(order_id: str) -> dict:
    """根据订单号查询订单状态。

    Args:
        order_id: 订单编号，如 ORD-20260803-001
    """
    # 真实实现里这里会查数据库
    return {"order_id": order_id, "status": "已发货", "eta": "2026-08-05"}


@mcp.resource("config://company-policy")
def company_policy() -> str:
    """把公司政策作为只读资源暴露给模型。"""
    return "退货政策：签收7日内无理由退货……"


if __name__ == "__main__":
    mcp.run()  # 以 stdio 方式启动，等待 MCP 客户端连接

# 把所有工具的描述建成向量索引（离线一次）
tool_index = SimpleVectorStore()
for t in ALL_TOOLS:  # 可能有上百个
    desc = f"{t['function']['name']}: {t['function']['description']}"
    tool_index.add(desc, metadata={"tool": t})

def select_tools(user_query: str, top_k: int = 5) -> list:
    """运行时按意图动态挑选最相关的少数工具。"""
    hits = tool_index.search(user_query, k=top_k)
    return [h.metadata["tool"] for h in hits]

# 每轮只把挑出来的 5 个工具的 Schema 发给模型
selected = select_tools("帮我把这个月的报销单导出成表格")
resp = client.chat.completions.create(model="gpt-4o", messages=msgs, tools=selected)

import time
import hashlib
from functools import wraps

# 已执行过的高危操作记录（幂等用），生产环境应落库/Redis
_executed_ops = {}


def with_retry(max_attempts=3, base_delay=1.0, timeout=10.0):
    """带指数退避重试 + 超时的工具执行装饰器。"""
    def deco(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            last_err = None
            for attempt in range(1, max_attempts + 1):
                try:
                    # 简化示意：真实项目用 signal/concurrent.futures 控制超时
                    return fn(*args, **kwargs)
                except (TimeoutError, ConnectionError) as e:
                    last_err = e
                    delay = base_delay * (2 ** (attempt - 1))  # 1s, 2s, 4s
                    time.sleep(delay)
            return {"error": f"工具执行失败（已重试{max_attempts}次）: {last_err}"}
        return wrapper
    return deco


def idempotent(fn):
    """对写操作做幂等保护：相同参数只真正执行一次。"""
    @wraps(fn)
    def wrapper(*args, **kwargs):
        key = hashlib.md5(
            (fn.__name__ + str(args) + str(kwargs)).encode()
        ).hexdigest()
        if key in _executed_ops:
            return _executed_ops[key]  # 直接返回上次结果，避免重复扣款/重复下单
        result = fn(*args, **kwargs)
        _executed_ops[key] = result
        return result
    return wrapper


HIGH_RISK = {"refund", "delete_account", "transfer_money"}


def execute_tool(name: str, args: dict, tools_impl: dict):
    """统一的工具执行入口：校验 + 高危拦截 + 执行。"""
    # 1) 参数校验：不信任模型输出
    if name not in tools_impl:
        return {"error": f"未知工具: {name}"}

    # 2) 高危操作插入人工确认关卡
    if name in HIGH_RISK:
        if not request_human_approval(name, args):
            return {"status": "cancelled", "reason": "用户未批准该操作"}

    # 3) 真正执行
    return tools_impl[name](**args)


@idempotent
@with_retry(max_attempts=3)
def transfer_money(to: str, amount: float) -> dict:
    # ... 真实转账逻辑 ...
    return {"status": "ok", "to": to, "amount": amount}