def dpo_loss(model, ref_model, batch, beta=0.1):
    # ① batch:chosen/rejected 两条序列,prompt 掩为 -100
    logps_chosen   = get_sequence_logprob(model, batch["chosen"])
    logps_rejected = get_sequence_logprob(model, batch["rejected"])
    with torch.no_grad():                ② 参考模型:预计算或在此计算
        ref_logps_chosen   = get_sequence_logprob(ref_model, batch["chosen"])
        ref_logps_rejected = get_sequence_logprob(ref_model, batch["rejected"])
    chosen_rewards   = beta * (logps_chosen - ref_logps_chosen)      ③ β 乘在 logp 差上
    rejected_rewards = beta * (logps_rejected - ref_logps_rejected)
    loss = -F.logsigmoid(chosen_rewards - rejected_rewards).mean()
    return loss

def get_sequence_logprob(model, sequences):
    """仅在回答 token 上求和的对数概率。"""
    outputs = model(sequences["input_ids"], attention_mask=sequences["mask"])
    logits = outputs.logits[:, :-1, :]   ④ 为 next-token 预测做位移
    labels = sequences["labels"][:, 1:]  # 位移后的标签
    log_probs = F.log_softmax(logits, dim=-1)
    token_logps = log_probs.gather(-1, labels.unsqueeze(-1)).squeeze(-1)
    mask = (labels != -100).float()      ⑤ 只对回答 token 求和
    return (token_logps * mask).sum(dim=-1)  # 形状:[batch_size]

from trl import SFTConfig, SFTTrainer, DataCollatorForCompletionOnlyLM
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.1-8B-Instruct")

def formatting_func(example):                    ① chat 模板拼接对话
    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": example["instruction"]},
        {"role": "assistant", "content": example["response"]}]
    return tokenizer.apply_chat_template(messages, tokenize=False)

response_template = "<|start_header_id|>assistant <|end_header_id|>"
collator = DataCollatorForCompletionOnlyLM(      ② 此 token 后才开始算 Loss
    response_template=response_template, tokenizer=tokenizer)

config = SFTConfig(
    max_seq_length=4096,
    packing=True,                                ③ 启用序列打包
    per_device_train_batch_size=4, learning_rate=2e-5)
trainer = SFTTrainer(model=model, args=config, train_dataset=dataset,
    data_collator=collator,                      ④ 仅对 completion 计算 Loss
    formatting_func=formatting_func)
trainer.train()

def route_query(query: str, llm) -> RouteDecision:  ① LLM 选源=一次「动作」
    prompt = f"""Given the query: "{query}"
Decide which knowledge source to use:
- vector_db: for internal documents, policies, past reports
- web_search: for current events, recent information
- sql_db: for numerical data, statistics, structured records
Return a JSON with: source, refined_query, reasoning."""
    return llm.with_structured_output(RouteDecision).invoke(prompt)

def retrieve_node(state: AgentState) -> AgentState:  ② 逐子查询扇出→融合
    new_docs = []
    for sq in state["sub_queries"]:
        new_docs.extend(retrieve_from_source(sq, route_query(sq)))
    return {**state, "retrieved_docs": new_docs,
            "iterations": state["iterations"] + 1}

def should_retrieve(state: AgentState) -> str:  ③ 充分?重检索?
    if state["context_sufficient"]:
        return "generate"
    if state["iterations"] >= state["max_iterations"]:
        return "generate"  # 预算耗尽:放弃检索,用现有内容生成
    return "retrieve"      # 换精炼子查询,回环再检索

class VectorMemoryStore:
    def write(self, content, importance=0.5, check_duplicates=True):
        if check_duplicates and self._is_duplicate(content):  ① 写入过滤
            return None   # 余弦相似度>0.95:近似重复,直接跳过
        if len(self.entries) >= self.max_entries:
            self._evict()  # 满:逐出「最不重要×最久未访」条目
        self.entries.append(MemoryEntry(content=content,
            embedding=self.embed_fn(content), importance=importance))

    def retrieve(self, query, k=5, recency_boost=True):
        dense_scores = self._cosine_similarities(self.embed_fn(query))
        for entry, d_score in zip(self.entries, dense_scores):  ② 相似×时近
            age_h = (datetime.now() - entry.timestamp).total_seconds()/3600
            recency = np.exp(-self.decay_rate * age_h)  # 每小时衰减
            score = (1 - self.recency_weight) * d_score \
                    + self.recency_weight * recency
        ...  # 组合分降序取 top-k;更新 access_count/last_accessed
class MemoryAugmentedAgent:
    def step(self, user_message: str) -> str:  ③ 记忆进 prompt
        memories = self.memory.retrieve(user_message, k=5)
        system = self.SYSTEM_PROMPT + f"\n\n{self._format_memories(memories)}"
        ...  # + get_hot_context() 热层;最近 6 轮 → llm → 执行记忆命令

class AgentHarness:
    def __init__(self, model, system_prompt, tools, approval_cb=None):
        self.executor = ToolExecutor(tools, approval_cb)  ① 工具分发表
        self.ctx_mgr = ContextManager(model, max_tokens=128_000)
        self.ctx_mgr.system_msg = Message(Role.SYSTEM, system_prompt)
    async def run(self, user_input: str) -> str:  # user 消息先入 context
        tool_defs = [t.to_api_dict() for t in self.tools]
        tool_tokens = sum(self.ctx_mgr.count_tokens(  ② 组装时计预算
            json.dumps(t)) for t in tool_defs)
        for iteration in range(self.MAX_ITERATIONS):  # 硬上限 50
            if not self.ctx_mgr.preflight_check(tool_tokens):
                return "I've run out of context space."  ③ 每轮预算闸
            response = await self.client.chat.completions.create(
                model=self.model, messages=self.ctx_mgr.build_messages(),
                tools=tool_defs, tool_choice="auto", temperature=0.0)
            choice = response.choices[0]
            if choice.finish_reason == "stop" or not choice.message.tool_calls:
                return choice.message.content or "Task complete."  ④ 终止
            results = await self._execute_tool_calls(choice.message.tool_calls)
            for tid, result in results.items():  ⑤ 观测回填
                self.ctx_mgr.add_message(Message(
                    Role.TOOL, result, tool_call_id=tid))

def agent_loop(goal: str, max_steps: int = 50,
               budget: TokenBudget = None):
    """The canonical agent loop structure."""
    state = initialize_state(goal)     ① 加载外部状态和目标
    for step in range(max_steps):
        thought = model.reason(state)  ② 推理(ReAct 风格)
        action = model.choose_action(state)
        observation = environment.execute(action)
        state = update_state(state, thought, action, observation)
        if state.token_count > CONTEXT_BUDGET * 0.8:
            state = compact(state)      汇总旧步骤
        if verifier.passes(state, goal): ③ 确定性检查
            persist_state(state, status="success")
            return Success(state)
        if no_progress_detected(state, window=3):  ④ 无进展
            persist_state(state, status="stuck")
            return escalate_to_human(state)
        if budget and budget.exhausted():
            persist_state(state, status="budget_exceeded")
            return escalate_to_human(state)
    persist_state(state, status="max_steps")   ⑤ 硬性上限
    return escalate_to_human(state)

from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("notes-server")          ① 一个实例=一个 server
NOTES_DIR = Path.home() / ".notes"

@mcp.tool()                               ② 声明式 tool
def create_note(title: str, content: str,
                tags: list[str] | None = None) -> str:
    """Create a new text note. Use this when
    the user wants to save information."""  ③ docstring 给 LLM 看
    safe = "".join(c if c.isalnum() or c in "-_" else "_"
                   for c in title).strip()  ④ 输入校验:防路径穿越
    p = NOTES_DIR / f"{safe}.md"
    p.write_text(f"---\ntitle: {title}\n---\n\n" + content,
                 encoding="utf-8")
    return f"Note saved to {p}"
@mcp.resource("notes://{title}")        ⑤ URI 路由的只读数据
def get_note(title: str) -> str:
    return (NOTES_DIR / f"{title}.md").read_text("utf-8")

if __name__ == "__main__":
    mcp.run()  ⑥ stdio / Streamable HTTP 自动选择

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    params = StdioServerParameters(          ① 如何拉起 server 子进程
        command="python", args=["notes_server.py"], env=None)
    async with stdio_client(params) as (read, write):
      async with ClientSession(read, write) as session:
        await session.initialize()            ② 能力协商握手

        tools = await session.list_tools()    ③ 工具发现
        for t in tools.tools:
            print(f" - {t.name}: {t.description[:60]}")

        result = await session.call_tool(      ④ 执行(先经用户同意)
            "create_note",
            arguments={"title": "MCP Architecture Notes",
                       "content": "MCP uses JSON-RPC 2.0.",
                       "tags": ["mcp"]})
        print(f"Tool result: {result.content[0].text}")

asyncio.run(main())

async def agent_stream(session_id: str, user_message: str):
    def sse(event_type: str, data: dict) -> str:
        return f"data: {json.dumps({'type': event_type, **data})}\n\n"
    approval_id = f"{session_id}_email_001"        ② Tier 3 审批门控
    approval_store[approval_id] = ev = asyncio.Event()
    yield sse("approval_required", {"approval_id": approval_id,
            "tool": "send_email", "tier": 3, "risk": "irreversible"})
    try:
        await asyncio.wait_for(ev.wait(), timeout=300)  ③ 挂起整条流
        ...  批准 → 执行 send_email;拒绝 → action_rejected
    except asyncio.TimeoutError:
        yield sse("approval_timeout", {"approval_id": approval_id})
    yield sse("token", {"content": "I've completed the research. "})
@app.get("/chat/stream")                     ① SSE 主通道
async def chat_stream(session_id: str, message: str):
    return StreamingResponse(agent_stream(session_id, message),
                             media_type="text/event-stream")
@app.post("/chat/approve")                    ④ 批准端点 → 恢复
async def handle_approval(req: ApprovalRequest):
    approval_results[req.approval_id] = {"approved": req.approved,
                                         "parameters": req.parameters}
    approval_store[req.approval_id].set()   set() 唤醒挂起的流