Embedding 与向量检索

Embedding 模型的工作原理

Embedding 模型将文本转换为高维向量，使得语义相似的文本在向量空间中距离相近。

"我喜欢吃苹果"   →  [0.12, -0.34, 0.56, ..., 0.78]  (1024维)
"我爱吃苹果"     →  [0.11, -0.33, 0.55, ..., 0.79]  ← 距离很近
"汽车维修指南"   →  [-0.78, 0.45, -0.23, ..., 0.01] ← 距离很远

Embedding 模型的训练通常包含**对比学习（Contrastive Learning）**环节：输入是许多已标记为相关/不相关的文本对，训练目标是让相关文本的向量相似度最大化，不相关的相似度最小化。

# 计算两个文本向量的余弦相似度
import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 示例
query_vec = embedding_model.encode("年假怎么申请？")
doc_vec = embedding_model.encode("员工年假申请流程")

similarity = cosine_similarity(query_vec, doc_vec)
print(f"相似度: {similarity:.4f}")  # 0.92 — 高度相关

向量数据库选型

向量数据库是 RAG 系统的核心基础设施。选择时需要权衡：

# 使用 Chroma 的示例（轻量级本地向量库）
import chromadb
from chromadb.utils import embedding_functions

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection(
    name="company_docs",
    embedding_function=embedding_functions.OpenAIEmbeddingFunction(
        api_key="your-api-key",
        model_name="text-embedding-3-small"
    )
)

# 添加文档
collection.add(
    documents=["员工年假申请流程...", "差旅报销标准..."],
    metadatas=[{"source": "hr_policy.pdf"}, {"source": "finance_policy.pdf"}],
    ids=["doc_1", "doc_2"]
)

# 检索
results = collection.query(
    query_texts=["年假怎么申请？"],
    n_results=3
)

文档切分策略

切分的基本矛盾

RAG 系统的检索效果高度依赖文档切分质量。切分的核心矛盾是：

切片太大 → 检索时引入过多噪声，模型注意力被稀释
切片太小 → 关键信息可能被截断，丢失上下文

不存在一种”放之四海皆准”的最优切分策略。你需要根据文档类型、检索场景和模型能力来选择。

五种主流切分方法

Token 切分

按固定 token 数量切分，适合需要精确控制 token 消耗的场景。

from llama_index.core.node_parser import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=256,     # 每个切片的 token 数
    chunk_overlap=30    # 相邻切片的重叠 token 数
)

nodes = splitter.get_nodes_from_documents(documents)

优点：精确控制上下文大小，适合较小上下文窗口的模型。 缺点：可能在句子中间截断，破坏语义完整性。

句子切分

保持句子完整性的切分方式，是大多数场景的默认选择。

from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=50
)

优点：保持自然语言的语义单元完整性。 缺点：对文档结构无感知，可能将相关段落拆分到不同切片。

句子窗口切片

在索引和检索时使用不同粒度：索引用小粒度便于精确匹配，检索返回时带上相邻的上下文窗口。

from llama_index.core.node_parser import SentenceWindowNodeParser

parser = SentenceWindowNodeParser(
    window_size=3,          # 检索时扩展 3 个相邻句子
    window_metadata_key="window",
    original_text_metadata_key="original"
)

核心优势：兼顾检索精度和上下文完整性。

语义切片

根据语义相关性自适应选择切分点，保持文档的语义连续性。

from llama_index.core.node_parser import SemanticSplitterNodeParser

splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,  # 相似度低于此阈值时切分
    embed_model=embed_model
)

适用场景：逻辑性好、内容专业的长文档。

Markdown 切片

专门针对 Markdown 结构化文档优化，按标题层级进行切分。

from llama_index.core.node_parser import MarkdownNodeParser

parser = MarkdownNodeParser()
# 自动识别 #、##、### 等标题层级，在每个标题段落处切分

最佳实践：将文档从 PDF/Word 转为 Markdown 后再进行切分，利用标题结构提升检索准确性。

切分策略选择指南

检索增强生成流水线

RAG 的两阶段架构

RAG（Retrieval-Augmented Generation）是解决大模型”知识不足”的核心架构。它将流程分为两个阶段：

阶段一：建立索引

RAG 两阶段流水线（索引 + 检索生成）— 来自 microsoft/generative-ai-for-beginners，MIT License

1. 文档解析：将 PDF、Word、Markdown 等格式解析为纯文本
2. 文本切片：按照选定的策略将文档切分为段落
3. 向量化：用 Embedding 模型将每个切片转为向量
4. 存储索引：将向量存入向量数据库，建立索引

阶段二：检索与生成

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ 用户提问  │ → │ 向量检索  │ → │ Prompt组装│ → │ 模型生成  │
│ Query     │    │ Retrieve  │    │ Augment   │    │ Generate  │
└──────────┘    └──────────┘    └──────────┘    └──────────┘

1. 用户提问：接收用户问题
2. 向量检索：将问题向量化，在向量数据库中检索最相似切片
3. Prompt 组装：将检索到的知识片段 + 原始问题 + 指令组装为完整 Prompt
4. 模型生成：大模型基于增强后的上下文生成回答

完整 RAG 流水线示例

from openai import OpenAI
import numpy as np

client = OpenAI()

class SimpleRAG:
    def __init__(self, embed_model="text-embedding-3-small"):
        self.embed_model = embed_model
        self.documents = []      # 存储文档文本
        self.embeddings = []     # 存储文档向量

    def add_documents(self, docs: list[str]):
        """建立索引：向量化并存储文档"""
        for doc in docs:
            vec = self._embed(doc)
            self.documents.append(doc)
            self.embeddings.append(vec)

    def _embed(self, text: str) -> np.ndarray:
        resp = client.embeddings.create(
            model=self.embed_model,
            input=text
        )
        return np.array(resp.data[0].embedding)

    def retrieve(self, query: str, top_k: int = 3) -> list[str]:
        """检索最相关的文档片段"""
        query_vec = self._embed(query)
        similarities = [
            np.dot(query_vec, doc_vec) /
            (np.linalg.norm(query_vec) * np.linalg.norm(doc_vec))
            for doc_vec in self.embeddings
        ]
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.documents[i] for i in top_indices]

    def query(self, question: str) -> str:
        """完整 RAG 查询"""
        contexts = self.retrieve(question)
        prompt = f"""请根据以下参考信息回答问题：

参考信息：
{' '.join(contexts)}

问题：{question}

如果参考信息不足以回答问题，请明确说明。"""

        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        return resp.choices[0].message.content

RAG 多轮对话中的问题改写

在 RAG 场景中实现多轮对话面临独特挑战。如果用户第二轮说”他的主管是谁？“，直接用这句话去检索会完全失败——系统不知道”他”指的是谁。

解决方案：问题改写（Query Rewriting）

def rewrite_query(conversation_history: list, current_query: str) -> str:
    """使用大模型将依赖上下文的问题改写为独立问题"""
    rewrite_prompt = f"""
    根据对话历史，将当前问题改写为不依赖上下文的独立问题。

    对话历史:
    {format_history(conversation_history)}

    当前问题: {current_query}

    改写后的问题:"""

    resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": rewrite_prompt}],
        temperature=0.1
    )
    return resp.choices[0].message.content

# 示例
# 历史: 用户问"张三的工位在哪里？" 助手答"在 A 座 5 层"
# 当前: "他的主管是谁？"
# 改写后: "张三的主管是谁？"

高级 RAG 模式

HyDE（Hypothetical Document Embeddings）

HyDE 的核心思想：先让模型”编造”一个假想答案，然后用这个假想答案去检索，而非用原始问题。这样做的直觉是——假想答案在语义上比问题更接近真实文档。

def hyde_retrieve(query: str, top_k: int = 3) -> list[str]:
    """使用 HyDE 方法进行检索"""
    # Step 1: 生成假想答案
    hyde_prompt = f"""
    Question: {query}
    Please write a passage that answers this question.
    Passage:"""

    hyde_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": hyde_prompt}]
    )
    hypothetical_doc = hyde_resp.choices[0].message.content

    # Step 2: 用假想答案而非原始问题进行检索
    query_vec = embed(hypothetical_doc)
    results = vector_db.search(query_vec, top_k=top_k)
    return results

HyDE 特别适合原始 query 很短但语义复杂的场景，因为假想答案能提供更多语义线索。

重排序（Re-Ranking）

初步向量检索（粗排）速度很快但精度有限。可以引入重排序模型（精排）对候选切片进行二次排序：

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')

def rerank(query: str, candidates: list[str], top_k: int = 3):
    """对候选切片进行重排序"""
    pairs = [(query, doc) for doc in candidates]
    scores = reranker.predict(pairs)

    # 按相关性分数排序
    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    return [doc for doc, _ in ranked[:top_k]]

检索策略优化全景

RAG 文档准备策略

构建高质量 RAG 系统的关键是文档准备。你需要理解”意图空间”和”知识空间”的关系：

Agentic RAG 架构 — 来自 microsoft/langchain-for-beginners，MIT License

核心原则：

• 在优化算法之前，先补充缺失的知识
• 在改善召回之前，先提升文档质量
• 持续收集用户意图，形成”数据采集-知识更新-专家验证”闭环

Function Calling 协议

AutoGen Studio — 微软的无代码多代理工作流构建器 — 来自 microsoft/autogen。在 Function Calling 与多 Agent 协作一节展示这些代表性工具。

DSPy — 用声明式方式编程（而非提示）LLM，Hermes 自进化管线使用它 — 来自 stanfordnlp/dspy

什么是 Function Calling

Function Calling（函数调用，也称 Tool Calling）是大模型 API 提供的标准能力。它允许模型在需要时输出结构化的工具调用指令，而不是纯文本回复。

工作流程如下：

Function Calling 协议流程（定义→决策→执行→回传）— 来自 bentoml/bentoml，Apache-2.0 License

JSON Schema 工具定义

# 定义工具列表
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "搜索公司内部知识库，获取政策文档和操作指南",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词或问题"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["hr", "it", "finance", "general"],
                        "description": "知识类别"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "发送邮件",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {
                        "type": "string",
                        "description": "收件人邮箱"
                    },
                    "subject": {
                        "type": "string",
                        "description": "邮件主题"
                    },
                    "body": {
                        "type": "string",
                        "description": "邮件正文"
                    }
                },
                "required": ["to", "subject", "body"]
            }
        }
    }
]

完整 Function Calling 循环

from openai import OpenAI
import json

client = OpenAI()

def execute_function_call(tool_call):
    """执行工具调用并返回结果"""
    func_name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    if func_name == "search_knowledge_base":
        # 实际的搜索逻辑
        result = knowledge_base.search(args["query"])
        return json.dumps(result)
    elif func_name == "send_email":
        # 实际的邮件发送逻辑
        result = email_service.send(
            to=args["to"],
            subject=args["subject"],
            body=args["body"]
        )
        return json.dumps({"status": "sent" if result else "failed"})
    else:
        return json.dumps({"error": f"Unknown function: {func_name}"})

def chat_with_tools(user_message: str, messages: list = None):
    """支持 Function Calling 的对话"""
    if messages is None:
        messages = [
            {"role": "system", "content": "你是一位企业助手，可以搜索知识库和发送邮件。"}
        ]

    messages.append({"role": "user", "content": user_message})

    # 第一次调用：模型决定是否使用工具
    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=tools
    )

    assistant_msg = response.choices[0].message

    # 如果模型要调用工具
    if assistant_msg.tool_calls:
        messages.append(assistant_msg)

        for tool_call in assistant_msg.tool_calls:
            # 执行工具
            result = execute_function_call(tool_call)
            # 将结果返回给模型
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # 第二次调用：模型基于工具结果生成最终回复
        final_response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )
        return final_response.choices[0].message.content

    # 如果模型直接回复
    return assistant_msg.content

工具定义最佳实践

1. 描述要精确：模型根据 description 判断何时调用工具，描述模糊会导致误调用
2. 参数要有约束：使用 enum、required、类型约束减少模型的参数错误
3. 函数职责单一：不要在一个函数中包办多种操作
4. 返回结构化结果：工具返回应易于模型理解，建议使用 JSON

# ❌ 糟糕的工具描述
{"name": "do_stuff", "description": "执行操作", "parameters": {...}}

# ✅ 精准的工具描述
{
    "name": "cancel_meeting",
    "description": "取消指定的会议，需要会议 ID 和取消原因",
    "parameters": {
        "properties": {
            "meeting_id": {"type": "string", "description": "会议的唯一标识"},
            "reason": {"type": "string", "description": "取消原因，将通知所有参会者"}
        },
        "required": ["meeting_id"]
    }
}

ReAct 推理-行动循环

ReAct 的核心思想

ReAct（Reasoning + Acting）是一种让模型交替进行**推理（Thought）和行动（Action）**的模式。它不是一次性生成最终答案，而是通过”思考→行动→观察→思考…”的循环来解决问题。

ReAct 框架：纯推理 / 纯行动 / 推理+行动的对比 — 来自 ReAct 论文作者项目页（ysymyth/ReAct），MIT License

一个典型的 ReAct 执行过程：

用户: "帮我查一下张三的部门，然后给他的主管发一封邮件"

Thought 1: 我需要先查找张三的部门信息
Action 1: search_knowledge_base(query="张三 部门")
Observation 1: "张三属于教研部，主管是李四([email protected])"

Thought 2: 已获得信息，现在需要写邮件给李四
Action 2: send_email(to="[email protected]", subject="关于张三",
                      body="...")
Observation 2: {"status": "sent"}

Thought 3: 任务完成
Final Answer: "已查到张三在教研部，并已向他的主管李四发送了邮件。"

手动实现 ReAct Agent

class ReActAgent:
    def __init__(self, tools: dict, max_iterations: int = 10):
        self.tools = tools
        self.max_iterations = max_iterations

    def run(self, task: str) -> str:
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": task}
        ]

        for i in range(self.max_iterations):
            response = client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                tools=self._format_tools()
            )

            msg = response.choices[0].message

            if msg.content and not msg.tool_calls:
                # 模型给出了最终答案
                return msg.content

            if msg.tool_calls:
                # 添加助手的工具调用到历史
                messages.append(msg)

                for tc in msg.tool_calls:
                    tool_name = tc.function.name
                    args = json.loads(tc.function.arguments)

                    # 执行工具
                    result = self.tools[tool_name](**args)

                    # 将观察结果添加到历史
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": json.dumps(result)
                    })
                    print(f"  [Tool: {tool_name}({args}) → {result}]")

        return "ReAct 循环达到最大迭代次数"

    def _build_system_prompt(self) -> str:
        return """你是一个能够使用工具的智能助手。
遵循 ReAct 模式：先思考，再行动，观察结果，然后决定下一步。
如果任务已完成，直接给出最终答案。"""

    def _format_tools(self) -> list:
        return [
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": func.__doc__ or "",
                    "parameters": get_schema(func)
                }
            }
            for name, func in self.tools.items()
        ]

ReAct 的优势与局限

优势：

• 可观察：每一步的思考和行动都有迹可循
• 可纠错：观察到错误结果后可以调整策略
• 可组合：自动将多个工具组合为解决方案

局限：

• 循环次数不可预测（可能陷入无限循环）
• 多次 API 调用增加延迟和成本
• 依赖工具返回的观察质量

MCP 协议与工具生态

为什么需要 MCP

Function Calling 有一个根本性问题：工具定义与消费耦合。每个 Agent 开发者都需要在自己的代码中硬编码工具的 JSON Schema。当工具 API 升级时，所有集成了该工具的 Agent 都需要手动更新。

MCP 生态架构：Host 应用经标准化协议连接各 MCP Server — 来自 modelcontextprotocol/modelcontextprotocol，CC-BY-4.0

MCP（Model Context Protocol） 的核心思想是”谁提供工具，谁定义工具”。将工具定义的职责从 Agent（消费方）转移到工具服务（提供方）。

MCP 的架构角色

构建 MCP Server 与 Client

MCP Server 示例：

from mcp.server import Server, stdio_server
from mcp.types import Tool, TextContent

app = Server("web-search")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="web_search",
            description="搜索互联网获取最新信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"},
                    "num_results": {"type": "integer", "default": 5}
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "web_search":
        results = search_engine.search(
            arguments["query"],
            num=arguments.get("num_results", 5)
        )
        return [TextContent(type="text", text=json.dumps(results))]
    raise ValueError(f"Unknown tool: {name}")

# 通过 stdio 启动 Server
async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

MCP Client 集成示例：

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

async def run_with_mcp_tools(user_query: str):
    server_params = StdioServerParameters(
        command="python",
        args=["web_search_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 拉取 MCP Server 的工具定义
            tools_result = await session.list_tools()
            tools = tools_result.tools

            # 转换为 OpenAI 格式的 tools 参数
            openai_tools = [
                {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema
                    }
                }
                for tool in tools
            ]

            # 标准 Function Calling 流程
            response = client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": user_query}],
                tools=openai_tools
            )

            # 如果模型要调用工具，通过 MCP Client 执行
            if response.choices[0].message.tool_calls:
                for tc in response.choices[0].message.tool_calls:
                    result = await session.call_tool(
                        tc.function.name,
                        json.loads(tc.function.arguments)
                    )
                    # ... 将结果返回给模型

MCP 的工程价值

• 解耦：工具定义和服务实现分离，各自独立迭代
• 动态发现：Agent 启动时自动拉取最新工具列表，零维护成本
• 生态效应：第三方可以提供标准化的 MCP Server，Agent 开发者只需集成 MCP Client
• 多传输协议：支持 stdio（本地进程通信）和 HTTP/SSE（远程通信）

反思与自纠正

为什么需要反思

大模型生成的内容不是每次都能用。它可能：

• 悄悄”修正”了代码中的变量名导致运行报错
• 基于错误的”事实”继续推理，产生级联错误
• 在一段很长的输出中忘了前面的约束

反思（Reflection）的核心思路是：让模型有机会审视和评估自己已经生成的完整内容，从而发现并修正错误。

自我反馈的两种模式

模式一：单步指令式反思

在一次调用中，通过 Prompt 指示模型生成答案的同时进行反思：

prompt_with_reflection = """
## 任务
1. 润色以下课程的语言表达，输出润色后的全文内容。
2. 反思润色后的内容：
   - 是否符合写作规范
   - 除了语言表达，其他内容是否被意外修改
   输出反思结果和修改建议。
3. 对照建议修改课程，输出修改后的全文内容。

## 课程初稿
{original_content}
"""

优点：实现简单，一次调用完成。 缺点：模型容易带相同思维偏差进行自我验证，陷入”自证正确”。

模式二：两步式”生成-反馈”

将生成和审查分离为两次独立调用：

def generate_and_review(content: str) -> str:
    # 第一步：生成
    draft_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "你是课程作家。润色以下内容，使其更具吸引力。"
        }, {
            "role": "user",
            "content": content
        }]
    )
    draft = draft_resp.choices[0].message.content

    # 第二步：审查（用不同的 system prompt！）
    review_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": """你是严苛的技术审查员。
请比对【原始内容】和【润色后内容】：
- 如果只修改了文案表达，代码等技术内容完全一致 → 回答"通过"
- 如果技术内容被修改 → 回答"不通过"，指出具体位置"""
        }, {
            "role": "user",
            "content": f"原始内容:\n{content}\n\n润色后内容:\n{draft}"
        }]
    )

    # 第三步：如果不通过，将审查结果反馈给写作 Agent 修正
    review = review_resp.choices[0].message.content
    if "不通过" in review:
        # ... 将反馈送回写作 Agent 修正
        pass

    return draft

核心优势：审查 Agent 的视角与写作 Agent 不同，避免角色偏见。甚至可以设置多个专项审查 Agent——事实审查、逻辑审查、风格审查、安全审查。

外部反馈

自我反馈有天然局限：模型无法验证内容在实际环境中的正确性。外部反馈 的思路是把生成结果放到真实环境中执行，用客观事实来验证。

def generate_code_and_validate(spec: str) -> str:
    # 第一步：生成代码
    code_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"根据以下需求写 Python 代码:\n{spec}"}]
    )
    code = extract_code(code_resp.choices[0].message.content)

    # 第二步：外部执行验证
    import subprocess, tempfile
    with tempfile.NamedTemporaryFile(suffix=".py", mode="w") as f:
        f.write(code)
        f.flush()
        result = subprocess.run(
            ["python", f.name],
            capture_output=True,
            text=True,
            timeout=30
        )

    # 第三步：将错误反馈给模型修正
    if result.returncode != 0:
        fix_prompt = f"""以下代码执行出错：

代码:
{code}

错误信息:
{result.stderr}

请修正代码并输出完整修正版本。"""
        fix_resp = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": fix_prompt}]
        )
        return fix_resp.choices[0].message.content

    return code

外部反馈的应用场景：

• 代码执行验证：用代码解释器运行代码，捕获运行时错误
• JSON Schema 校验：用 Pydantic 等库验证结构化输出
• 数值计算验证：用计算器工具验证数学结果
• 可视化渲染验证：生成图表后让模型”看”渲染结果进行视觉检查

Plan & Execute 模式

为什么需要显式规划

直接让 Agent 执行复杂任务时，常见的问题是：

• 遗忘：在处理到后面步骤时”忘记”了前面的约束
• 级联错误：早期错误成为后续推理的基础，错误指数放大
• 结构错失：模型倾向于线性处理，无法识别任务的并行/依赖关系

Plan & Execute 模式的核心思想：先规划再执行——先制定完整的行动计划，通过审查确认后再逐步执行。

Plan Mode 实现

from typing import List
from pydantic import BaseModel

class PlanStep(BaseModel):
    step_id: int
    description: str
    dependencies: List[int] = []  # 依赖的步骤 ID
    tool: str = ""               # 使用的工具
    expected_output: str = ""    # 预期产出

class ExecutionPlan(BaseModel):
    goal: str
    steps: List[PlanStep]

def plan_and_execute(task: str) -> str:
    # Phase 1: 制定计划
    plan_prompt = f"""
    你是一位项目规划专家。请为以下任务制定详细的执行计划。

    要求：
    1. 将任务分解为具体步骤
    2. 标注步骤之间的依赖关系
    3. 每个步骤说明预期产出

    任务: {task}

    请以 JSON 格式输出计划。"""

    plan_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": plan_prompt}],
        response_format={"type": "json_object"}
    )
    plan = ExecutionPlan.model_validate_json(
        plan_resp.choices[0].message.content
    )

    # Phase 2: 按依赖关系执行
    results = {}
    executed = set()

    while len(executed) < len(plan.steps):
        for step in plan.steps:
            if step.step_id in executed:
                continue
            # 检查依赖是否都已执行
            if all(dep in executed for dep in step.dependencies):
                # 执行步骤
                result = execute_step(step, results)
                results[step.step_id] = result
                executed.add(step.step_id)

    # Phase 3: 汇总结果
    return summarize_results(plan, results)

固化工作流：Pipeline 模式

当任务的步骤是确定的、可重复的，应该将其固化为流水线：

Input → Step 1 → Step 2 → Step 3 → ... → Output

class Pipeline:
    """固定流水线：每个步骤的输出是下一步的输入"""

    def __init__(self):
        self.steps = []

    def add_step(self, name: str, func):
        self.steps.append({"name": name, "func": func})

    def run(self, input_data):
        result = input_data
        for step in self.steps:
            print(f"  [执行] {step['name']}")
            result = step["func"](result)
        return result

# 示例：文档处理流水线
pipeline = Pipeline()
pipeline.add_step("解析 PDF", parse_pdf_to_text)
pipeline.add_step("文本清洗", clean_text)
pipeline.add_step("分段", split_sections)
pipeline.add_step("向量化", vectorize_chunks)
pipeline.add_step("存储索引", store_to_vectordb)

pipeline.run("document.pdf")

工作流编排模式

五种核心工作流模式

复杂任务需要将 Agent 节点按特定拓扑组织。以下是五种核心模式：

分支选择（Branching/Router）

在入口节点判断任务类型，分流到不同处理路径。

def router_agent(user_input: str):
    """根据意图分流到不同的处理流水线"""
    classify_prompt = f"""
    分析以下用户请求的类型，只回复一个词:
    - code_review: 检查/验证代码
    - style_review: 润色/优化语言
    - fact_check: 验证事实/概念准确性

    请求: {user_input}
    类型:"""

    intent = client.chat.completions.create(
        model="gpt-4o-mini",  # 用轻量模型节省成本
        messages=[{"role": "user", "content": classify_prompt}],
        temperature=0
    ).choices[0].message.content.strip()

    pipelines = {
        "code_review": code_review_pipeline,
        "style_review": style_review_pipeline,
        "fact_check": fact_check_pipeline
    }
    return pipelines.get(intent, default_pipeline)(user_input)

并行执行（Parallel）

将互不依赖的子任务同时分发，最后汇总。

              ┌→ 代码检查 ──┐
用户输入 → Split ─┼→ 事实审查 ──┼→ Merge → 汇总报告
              └→ 风格审查 ──┘

import asyncio

async def parallel_review(notebook_content: str):
    """对课程内容并行进行三项审查"""
    tasks = [
        asyncio.create_task(check_code(notebook_content)),
        asyncio.create_task(check_facts(notebook_content)),
        asyncio.create_task(check_style(notebook_content))
    ]

    code_result, fact_result, style_result = await asyncio.gather(*tasks)

    # 汇总
    return generate_summary_report(code_result, fact_result, style_result)

混合专家（Mixture-of-Agents, MoA）

多个不同模型处理同一任务，由聚合器综合最优结果。

             ┌→ 模型 A (擅长推理) ──┐
用户问题 → Split ─┼→ 模型 B (擅长创意) ──┼→ Aggregator → 最优答案
             └→ 模型 C (擅长准确性) ─┘

MoA 的核心发现是模型的”协作性”（Collaborativeness）：当一个模型能参考其他模型的输出时，往往能生成质量更高的响应。

def mixture_of_agents(task: str):
    """MOA 实现：多模型 + 聚合"""
    # 第一层：提议者并行生成
    proposers = ["gpt-4", "claude-3-opus", "gemini-pro"]
    proposals = []

    for model in proposers:
        resp = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": task}]
        )
        proposals.append(resp.choices[0].message.content)

    # 第二层：聚合器综合
    aggregator_prompt = f"""
    以下是对同一问题的 {len(proposals)} 份回答。请综合它们的优点，
    生成一份最优答案。

    {format_proposals(proposals)}

    综合答案:"""

    final = client.chat.completions.create(
        model="gpt-4",  # 使用最强模型做聚合
        messages=[{"role": "user", "content": aggregator_prompt}]
    )
    return final.choices[0].message.content

人机协作（Human-in-the-Loop, HITL）

在关键节点引入人工审核，形成”AI 执行 → 人工审批 → AI 继续”的循环。

def hitl_workflow(task: str):
    """人机协作工作流"""
    plan = generate_plan(task)
    print(f"执行计划:\n{format_plan(plan)}")

    approval = input("是否批准此计划？(y/n): ")
    if approval.lower() != 'y':
        return "任务已取消"

    for step in plan.steps:
        result = execute_step(step)
        print(f"步骤 {step.step_id} 完成: {result['summary']}")

        if step.get("requires_review"):
            review = input(f"请审核步骤结果 (approve/modify/reject): ")
            if review == "reject":
                print("步骤被拒绝，重新执行...")
                result = execute_step(step, feedback=review)

    return generate_final_output()

模式选择方法论

最佳实践是”探索-固化”混合模式：先用 Plan & Execute 探索出最优方案，再固化为 Pipeline 用于大规模生产。

层级协作模式

Leader-Worker 架构

层级协作（Hierarchical/Team Leader Pattern）是最直观的多 Agent 协作模式。它模拟了”项目经理 + 团队成员”的组织结构：

Leader Agent 负责：

1. 接收和理解顶层任务
2. 分解为子任务并分派给合适的 Worker
3. 跟踪整体进度
4. 汇总 Worker 成果

Worker Agent 各自拥有特定领域的专长，专注于执行分配到的子任务。

通过 Handoff 实现层级协作

class LeaderWorkerSystem:
    """层级协作系统实现"""

    def __init__(self):
        self.workers = {
            "instructional_designer": self._create_worker(
                "你是教学设计师，擅长设计课程大纲和学习路径。"
            ),
            "data_scientist": self._create_worker(
                "你是数据科学家，擅长编写 Python 数据分析代码和案例。"
            ),
            "content_writer": self._create_worker(
                "你是内容编写者，擅长将技术内容转化为生动的课程文稿。"
            )
        }
        self.leader = self._create_leader()

    def _create_leader(self):
        return {
            "system_prompt": """你是课程项目主管。你的职责是：
1. 分析需求，分解为子任务
2. 将任务分派给合适的专家
3. 整合各专家的成果为完整课程
可用的专家团队：instructional_designer, data_scientist, content_writer""",
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": "delegate_to_worker",
                        "description": "将子任务分派给指定的专家",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "worker": {
                                    "type": "string",
                                    "enum": list(self.workers.keys()),
                                    "description": "接收任务的专家"
                                },
                                "task": {
                                    "type": "string",
                                    "description": "具体的子任务描述"
                                }
                            },
                            "required": ["worker", "task"]
                        }
                    }
                }
            ]
        }

    def run(self, project_brief: str) -> str:
        """执行一个完整的课程开发项目"""
        messages = [
            {"role": "system", "content": self.leader["system_prompt"]},
            {"role": "user", "content": project_brief}
        ]

        # Leader 循环
        while True:
            response = client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                tools=self.leader["tools"]
            )
            msg = response.choices[0].message

            if msg.content and not msg.tool_calls:
                return msg.content  # 最终汇总输出

            if msg.tool_calls:
                messages.append(msg)
                for tc in msg.tool_calls:
                    if tc.function.name == "delegate_to_worker":
                        args = json.loads(tc.function.arguments)
                        # 调用 Worker 执行子任务
                        worker_result = self._run_worker(
                            args["worker"], args["task"]
                        )
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tc.id,
                            "content": worker_result
                        })

    def _run_worker(self, worker_name: str, task: str) -> str:
        worker = self.workers[worker_name]
        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": worker},
                {"role": "user", "content": task}
            ]
        )
        return resp.choices[0].message.content

层级协作的优缺点

优点：

• 结构清晰，每个 Agent 职责明确
• Leader 把控全局，不会偏离目标
• 每个 Worker 拥有独立上下文窗口，更聚焦
• 支持并行分派任务

缺点：

• Worker 之间不直接沟通，信息传递有延迟/失真
• Leader 成为单点瓶颈
• 各模块组合后可能缺乏整体流畅感

黑板协作模式

去中心化的共创

黑板模式（Blackboard/Co-creation Pattern）模拟了”专家围坐白板头脑风暴”的工作方式。它没有中心化的协调者，所有 Agent 平等地读写共享空间：

黑板模式实现

class BlackboardSystem:
    """黑板协作系统"""

    def __init__(self, agents: dict, max_rounds: int = 3):
        self.agents = agents
        self.max_rounds = max_rounds
        self.blackboard = []  # 共享空间

    def run(self, problem: str) -> str:
        # 将问题写入黑板
        self.blackboard.append({"source": "user", "content": problem})

        for round_num in range(self.max_rounds):
            print(f"\n=== 第 {round_num + 1} 轮 ===")
            new_contributions = []

            # 所有 Agent 并行读取黑板并贡献
            for name, agent_config in self.agents.items():
                contribution = self._agent_contribute(
                    name, agent_config, self.blackboard
                )
                if contribution:
                    new_contributions.append({
                        "source": name,
                        "content": contribution
                    })

            # 将新贡献写入黑板
            self.blackboard.extend(new_contributions)

            # 检查是否达成共识
            if self._check_consensus():
                break

        return self._synthesize_final_answer()

    def _agent_contribute(self, name, config, blackboard):
        """每个 Agent 读取黑板后贡献自己的想法"""
        board_text = self._format_blackboard(blackboard)

        prompt = f"""你是一位{config['role']}。

当前共享黑板上的内容:
{board_text}

请基于现有讨论，提出你的见解、补充、质疑或新想法。
如果已有方案已经完善，你可以表示同意并说明理由。"""

        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": config["system_prompt"]},
                {"role": "user", "content": prompt}
            ]
        )
        return resp.choices[0].message.content

    def _format_blackboard(self, blackboard):
        return "\n\n".join([
            f"[{entry['source']}]: {entry['content']}"
            for entry in blackboard
        ])

    def _check_consensus(self):
        """检查黑板上的最新贡献是否已形成共识"""
        # 实现共识检测逻辑
        pass

    def _synthesize_final_answer(self):
        """从黑板内容中合成最终方案"""
        pass

黑板模式 vs 层级模式

协作模式选择方法论

从现实世界学习

优秀的多 Agent 系统设计来源于对现实世界团队协作的观察和提炼。与其记忆抽象的模式名称，不如走进业务，观察人类专家团队如何完成类似任务。

观察的三个维度：

混合模式设计

实际项目中很少使用单一模式。常见的是混合设计：

                    ┌─────────────┐
                    │   Leader    │  ← 层级模式
                    └──────┬──────┘
           ┌───────────────┼───────────────┐
           ▼               ▼               ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ 教学设计师 │    │ 内容编写者 │    │ 审核 Leader│
    │  Worker   │    │  Worker   │    └─────┬────┘
    └──────────┘    └──────────┘           │
                                  ┌────────┼────────┐
                                  ▼        ▼        ▼
                             ┌──────┐ ┌──────┐ ┌──────┐
                             │代码检 │ │事实检 │ │风格检 │  ← 并行模式
                             └──────┘ └──────┘ └──────┘
                                 │        │        │
                                 └────────┼────────┘
                                          ▼
                                    ┌──────────┐
                                    │ 综合报告  │
                                    └──────────┘

成本意识

多 Agent 系统的 Token 消耗通常是单 Agent 的 3-5 倍。设计时需要权衡：

def estimate_cost(num_agents: int, avg_tokens_per_agent: int,
                  rounds: int = 1, price_per_1k: float = 0.01):
    """估算多 Agent 系统的 Token 成本"""
    total_tokens = num_agents * avg_tokens_per_agent * rounds
    return total_tokens * price_per_1k / 1000

# 示例：5 个 Agent，每个 2000 tokens，3 轮黑板讨论
cost = estimate_cost(5, 2000, 3)
print(f"预估成本: ${cost:.2f}")
# 实际数字可能更高，因为还要包含黑板内容的重复传输

短期记忆管理

无状态性：问题的根源

大语言模型本质上是**无状态（Stateless）**的。每次 API 调用都是独立的——它不会记得上一轮对话的内容、你的偏好、或者之前达成的共识。

# 这两次调用彼此完全独立
response1 = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "我叫张三"}]
)
# 回复: "你好张三！有什么可以帮助你的？"

response2 = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "我叫什么名字？"}]
)
# 回复: "抱歉，我不知道你的名字，因为我们之前没有对话过。"

解决方案：维护对话历史列表，每次调用时发送完整历史。

class ConversationBuffer:
    """最简单的短期记忆：保存完整对话历史"""

    def __init__(self, system_prompt: str = ""):
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})

    def chat(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        response = client.chat.completions.create(
            model="gpt-4",
            messages=self.messages
        )
        reply = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": reply})
        return reply

上下文窗口压力

随着对话轮次增加，完整历史方案面临三个致命问题：

1. 超出上下文窗口：历史长度超过模型限制 → 程序报错
2. 成本失控：每次调用都重新发送全部历史 → Token 消耗线性增长
3. 注意力稀释：长上下文中，模型对中间部分的信息处理能力显著下降

三种记忆管理策略

策略一：固定窗口截断（Context Truncation）

只保留最近的 N 轮对话或 N 个 Token。

class TruncationMemory:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self.messages = []

    def add_and_truncate(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        # 从最旧的消息开始删除，直到总 token 数在限制内
        while self._total_tokens() > self.max_tokens:
            self.messages.pop(0)  # 删除最旧的非 system 消息

    def _total_tokens(self):
        return sum(count_tokens(m["content"]) for m in self.messages)

优点：实现极其简单，计算开销小。 缺点：如果关键信息在早期对话中，被截断后 Agent 就会”失忆”。

策略二：滚动摘要（Rolling Summary）

在遗忘之前，先把重点提炼出来。

对话历史: [msg1, msg2, msg3, msg4, msg5, msg6, msg7, msg8]
                          ↓ 前半部分压缩
         [摘要(m1-m4), msg5, msg6, msg7, msg8]
                          ↓ 继续压缩
         [摘要(m1-m6), msg7, msg8]

class RollingSummaryMemory:
    def __init__(self, summary_trigger_tokens: int = 3000):
        self.summary_trigger = summary_trigger_tokens
        self.messages = []
        self.summary = ""

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        if self._total_tokens() > self.summary_trigger:
            self._compress()

    def _compress(self):
        """压缩前半部分对话为摘要"""
        split_point = len(self.messages) // 2
        to_compress = self.messages[:split_point]
        remaining = self.messages[split_point:]

        compress_prompt = f"""
        请将以下对话历史总结为一段简洁的摘要，保留关键信息：

        对话:
        {format_messages(to_compress)}

        摘要:"""

        resp = client.chat.completions.create(
            model="gpt-4o-mini",  # 用轻量模型摘要
            messages=[{"role": "user", "content": compress_prompt}]
        )
        self.summary = resp.choices[0].message.content

        # 用摘要替换被压缩的消息
        self.messages = [
            {"role": "system", "content": f"对话历史摘要:\n{self.summary}"}
        ] + remaining

    def _total_tokens(self):
        return sum(count_tokens(m["content"]) for m in self.messages)

优点：在压缩长度的同时保留核心信息，维持长期连贯性。 缺点：额外 API 调用成本，摘要质量直接影响后续对话。

策略三：向量化召回（Vector-based Retrieval）

最智能的方式：将对话历史存入向量数据库，按需检索最相关的记忆。

class VectorBasedMemory:
    def __init__(self):
        self.conversations = []  # 完整对话记录
        self.embeddings = []     # 每轮对话的向量
        self.embed_model = "text-embedding-3-small"

    def store_conversation(self, user_msg: str, assistant_msg: str):
        """存储一轮对话并向量化"""
        conversation_text = f"User: {user_msg}\nAssistant: {assistant_msg}"
        self.conversations.append(conversation_text)

        vec = client.embeddings.create(
            model=self.embed_model,
            input=conversation_text
        )
        self.embeddings.append(vec.data[0].embedding)

    def retrieve_relevant(self, current_query: str, top_k: int = 5):
        """检索与当前问题最相关的历史对话"""
        query_vec = client.embeddings.create(
            model=self.embed_model,
            input=current_query
        ).data[0].embedding

        # 计算相似度
        similarities = [
            np.dot(query_vec, mem_vec) /
            (np.linalg.norm(query_vec) * np.linalg.norm(mem_vec))
            for mem_vec in self.embeddings
        ]

        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.conversations[i] for i in top_indices]

优点：根本摆脱上下文窗口的长度限制，语义精准匹配。 缺点：系统复杂度最高，引入 Embedding 模型和向量数据库。

策略选择指南

长期记忆与向量存储

从被动上下文到主动记忆管理

真正智能的 Agent 不应只是被动地接收处理好的上下文，它应该能主动管理自己的记忆——自己决定什么时候该记住什么，什么时候该回忆起什么。

这需要为 Agent 提供两个核心工具：

# Agent 可用的记忆管理工具
memory_tools = [
    {
        "type": "function",
        "function": {
            "name": "record_to_memory",
            "description": "将重要信息存入长期记忆。当用户明确表达偏好、提供关键信息或完成重要决策时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "content": {
                        "type": "string",
                        "description": "需要记忆的内容"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["preference", "fact", "decision", "context"],
                        "description": "记忆类别"
                    }
                },
                "required": ["content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "retrieve_from_memory",
            "description": "从长期记忆中检索相关信息。在需要回忆用户偏好、历史决策或之前讨论过的内容时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "检索查询"
                    },
                    "category": {
                        "type": "string",
                        "description": "可选，限定检索的记忆类别"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

短期记忆 vs 长期记忆

┌─────────────────────────────────────────────────────┐
│                    记忆系统架构                        │
├─────────────────────────────────────────────────────┤
│                                                     │
│  短期记忆 (Short-term Memory)                         │
│  ┌───────────────────────────────────────────────┐  │
│  │ 存储：对话缓冲区 (messages list)                │  │
│  │ 管理：截断 / 摘要                              │  │
│  │ 生命周期：当前会话                              │  │
│  │ 职责：维持会话连贯性，记住"刚刚在聊什么"          │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  长期记忆 (Long-term Memory)                          │
│  ┌───────────────────────────────────────────────┐  │
│  │ 存储：向量数据库 + 元数据存储                   │  │
│  │ 管理：向量化召回 + 工具化调用                   │  │
│  │ 生命周期：跨会话                                │  │
│  │ 职责：持久化关键信息，支持跨会话智能检索          │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
└─────────────────────────────────────────────────────┘

记忆管理最佳实践

1. 有选择地记忆：记忆并非越多越好。低价值信息会干扰后续检索。建立写入准入机制——仅当用户显式要求或信息重要性高于阈值时才写入。

2. 持续治理：记忆是动态数据资产。定期清理过时信息、合并重复条目、校验事实准确性。提供用户管理记忆的接口（查看、修改、删除）。

3. 场景化应用：不同场景对记忆需求不同。课程文档工作流中不应记录个性化偏好；对产品事实信息（API 参数、功能限制等）则应记录并定期审查有效性。

Skill 系统设计

从 Prompt 到 Skill 的演进

一段精心编写的 Prompt 价值很高，但它只在当前会话中生效。会话结束，知识就散了。Skill 是将 Prompt 升级为可复用、可版本管理、可团队共享的专业知识模块。

演进路径：

临时 Prompt ("帮我审核一下这份课程")
    ↓ 问题：每次都要重新写，标准不一致
固定 Prompt (存在聊天记录里)
    ↓ 问题：散落、难找、无法协作
独立文件 (course-review.md)
    ↓ 问题：文件膨胀，难以维护
知识库目录 (course-review/)
    ↓ 问题：仍需手动告诉 Agent 该做什么
Skill (SKILL.md + 资源文件 + 脚本)

Skill 的结构

一个规范的 Skill 由 YAML frontmatter + Markdown body 组成：

---
name: course-review
description: |
  审查课程内容的技术准确性、代码正确性和教学质量。
  当用户要求审查、审计或评估现有的课程或培训材料时使用此技能。
---

# 课程审查技能

## 审查流程
1. 提取 Notebook 目录结构，了解整体章节编排
2. 按章节逐段检查，对照以下方向：
   - 代码可执行性（详见 [code-quality.md](code-quality.md)）
   - 内容准确性（详见 [content-accuracy.md](content-accuracy.md)）
   - 讲解风格（详见 [style-guide.md](style-guide.md)）
   - 过时 API（详见 [outdated-api.md](outdated-api.md)）
3. 汇总审查结果，按输出格式生成报告

## 反模式清单
- ❌ 不要修改代码中的变量名、API版本号
- ❌ 不要跳过任何一项检查
- ❌ 不要在审核过程中引入新的技术概念

## 输出格式
每项检查给出：通过/不通过/需人工复核 + 位置 + 修改建议

目录结构：

course-review/
├── SKILL.md              # 主指令入口
├── code-quality.md       # 代码可执行性检查项
├── content-accuracy.md   # 事实准确性检查项
├── style-guide.md        # 讲解风格正例和反例
├── outdated-api.md       # 过时 API 对照表
└── scripts/
    ├── extract_toc.py    # 提取 Notebook 目录
    └── validate_code.py  # 自动执行代码验证

Skill vs RAG

很多人会混淆 Skill 和 RAG。它们的核心区别是：

编写高质量 Skill 的五步法

第一步：判断值不值得做
  ├─ 这个任务有"专家直觉"吗？（专家能做好但新手容易漏边界条件）
  ├─ 这个任务足够复杂吗？（3步以内GUI能完成的，不做）
  └─ 这个任务会反复执行吗？（只做一次，不做）

第二步：提取该写什么
  ├─ 提取专家的决策树而非单纯的步骤
  ├─ 注入反模式检查（"哪些坑绝对不能踩"）
  ├─ Template 模式：提供标准化的输出模板
  └─ Examples 模式：用示例代替文字描述

第三步：写好指令
  ├─ 简洁：每句话都要值得它的 token 成本
  ├─ 自由度匹配：约束程度匹配任务风险
  │   ├─ 低自由度（数据库迁移）：精确脚本
  │   ├─ 中自由度（报告生成）：伪代码/参数化
  │   └─ 高自由度（代码审查）：文本指令
  └─ 渐进式披露：主文件保持精简，详情按需加载

第四步：配好工具
  ├─ 工作流：可追踪的 Checklist
  ├─ 反馈循环：运行→检查→修复→重复
  ├─ 高风险操作：先验证计划再执行
  └─ AI 友好脚本：结构化状态 + 修复线索 + 优雅降级 + 幂等安全

第五步：验证和迭代
  ├─ 阶段一：建立评测基线（先有评测再写 Skill）
  ├─ 阶段二：提取 Skill（用 AI 总结反复提供的信息）
  └─ 阶段三：双 Agent 测试迭代（设计者 vs 使用者）

Skill as Code

将 Skill 视为代码，享受代码工程的方法论：

• 版本管理：Skill 目录纳入 Git，每次修改都有历史记录
• 代码审查：新增/修改 Skill 需要 PR 审查
• CI/CD：Skill 变更触发评测流水线，确保不退化
• 社区共享：像开源库一样共享和复用 Skill

渐进式信息披露与 Skill as Code

渐进式信息披露的设计哲学

传统的 Prompt 工程面临的核心矛盾：信息太多 → 上下文拥挤、注意力稀释；信息太少 → Agent 缺乏足够的知识支撑决策。

渐进式信息披露（Progressive Disclosure） 是一种解决此矛盾的设计模式：

第 1 层：Skill 列表（Agent 启动时可见）
  ↓ Agent 选择激活某个 Skill
第 2 层：SKILL.md（Skill 激活后加载）
  ↓ Agent 执行到某一步骤
第 3 层：子文件（按需加载对应资源）
  ↓ 需要验证某个技术细节
第 4 层：脚本执行（提供确定性结果）

设计原则：

• 保持扁平结构，避免深层嵌套引用
• SKILL.md 直接链接所有资源文件，确保”一步”可达
• 不要把所有内容塞进 SKILL.md——它只是入口

AI 友好脚本设计

工具脚本的输出质量直接影响 Agent 的表现。四个关键原则：

1. 结构化状态反馈——输出 JSON 而非自由文本：

# ❌ 差的输出
print("Error: exit code 1")
print("Could not find module 'openpyxl'")

# ✅ AI 友好的输出
print(json.dumps({
    "status": "failed",
    "error_code": "MODULE_NOT_FOUND",
    "missing_module": "openpyxl",
    "fix_hint": "运行 pip install openpyxl 安装缺失的依赖",
    "fallback_available": True,
    "affected_cells": [42, 43, 44]
}))

2. 错误信息包含修复线索——告诉 Agent “怎么修”而不只是”出了什么错”。

3. 优雅降级而非崩溃——在可能的情况下提供默认值并继续。

4. 幂等与安全——支持重复执行而不产生副作用：

评测框架设计

为什么”感觉还行”靠不住

Agent 的优化如果只凭主观感觉，会导致：

• 难以量化：“感觉更好”无法作为工程决策依据
• 缺乏标准：不同的测试者、不同时间点评测标准可能漂移
• 无法复现：无法系统性回归测试，确保新改动没有破坏旧功能

评测驱动开发（Evaluation-Driven Development） 将评测从开发流程的末端提升到核心位置：

端到端评测（End-to-End Evaluation）

端到端评测关注最终输出，回答”这个 Agent 对用户来说好用吗？”。

评测指标分为两类：

白盒化评测（White-box Evaluation）

当 Agent 流程变复杂时，端到端评测无法定位具体问题。白盒化评测主张深入系统内部，为关键组件单独设计评测体系。

端到端评测 (只看最终输出):
  Agent 整体 → 得分 4.2/5 → 但不知道哪里出了问题

白盒化评测 (检查中间环节):
  ┌────────────┐    ┌────────────┐    ┌────────────┐
  │ 概念解释    │    │ 代码生成    │    │ 风格调整    │
  │ 得分 3.1/5 │    │ 得分 4.8/5 │    │ 得分 4.5/5 │
  └────────────┘    └────────────┘    └────────────┘
       ↑ 瓶颈在这里！

白盒化评测的核心优势：

• 清晰信号：无干扰的改进信号，聚焦真正瓶颈
• 快速迭代：只需测试单个组件，无需运行整个流程
• 精准优化：每个改动效果可精确度量

LLM-as-Judge

用大模型做评测员

让另一个大模型扮演”评测专家”，根据定义的指标和评分细则自动打分。

def llm_judge_evaluate(generated_output: str, criteria: dict):
    """使用 LLM 作为评测员"""
    judge_prompt = f"""
    你是一位专业的课程质量评测员。请按照以下标准评分：

    评测标准:
    {json.dumps(criteria, indent=2, ensure_ascii=False)}

    待评测内容:
    {generated_output}

    请输出 JSON 格式的评分结果：
    {{
        "scores": {{
            "accuracy": <1-5>,
            "clarity": <1-5>,
            "engagement": <1-5>
        }},
        "overall": <1-5>,
        "comments": "综合评价"
    }}"""

    resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": judge_prompt}],
        response_format={"type": "json_object"}
    )
    return json.loads(resp.choices[0].message.content)

LLM 评测员的偏见

使用 LLM 作为评测员时，必须警惕其固有偏见：

最佳实践：在初期使用人类专家建立”黄金测试集”，用它来校准 LLM 评测员。定期用人工抽检验证自动评测的一致性。

评测指标拆解

将模糊的评测目标拆解为可逐一检查的具体细则：

# 内容质量评测拆解示例
评测维度: "内容质量"
细则:
  - id: "pain_point"
    description: "是否以一个具体的痛点作为开篇？"
    type: "boolean"
  - id: "theory_depth"
    description: "是否清晰指出了初步解法的局限并引出核心理论？"
    type: "boolean"
  - id: "code_relevance"
    description: "代码示例与讲解的理论是否紧密相关且足够简化？"
    type: "boolean"
  - id: "anti_pattern_check"
    description: "是否避免了'恭喜你解锁新技能'之类的俏皮话？"
    type: "boolean"

评测驱动迭代

评测闭环

评测不是一次性的，而是持续驱动改进的引擎：

    ┌──────────────────────────────────┐
    │                                  │
    ▼                                  │
┌─────────┐   ┌──────────┐   ┌─────────┐
│ 构建 MVP │ → │ 发现问题  │ → │ 提炼指标 │
└─────────┘   └──────────┘   └─────────┘
                                  │
                                  ▼
┌─────────┐   ┌──────────┐   ┌─────────┐
│ 部署上线 │ ← │ 回归测试  │ ← │ 优化改进 │
└─────────┘   └──────────┘   └─────────┘
                                  │
                                  └────→ (循环)

业务专家主导评测标准

评测指标（特别是主观指标）必须由最资深的业务专家主导制定：

1. 用业务目标动员参与：不说”帮我们定义评测指标”，而说”这个 Agent 要帮你把课程制作周期从2周缩短到3天，同时保持90%以上的用户满意度”。

2. 提供结构化工具降低门槛：评分量表模板、案例标注工具、“如果你只能看三个指标就判断课程好坏，你会选哪三个？” 这种引导性问题。

3. 建立持续协作机制：每周回顾会上，专家看数据、技术团队调系统，共同决策。

评测的效率杠杆

并非所有评测都需要全自动、全覆盖。从最简单的方法开始：

Level 1: 手动抽查 → "复制代码跑一下"
Level 2: 自动化脚本 → "写个脚本批量执行"
Level 3: 集成评测流水线 → CI/CD 集成，每次 PR 自动触发
Level 4: 持续监控 → 生产环境实时监测关键指标

每个 Level 的投入产出比不同。在初期，Level 1 的投入产出比最高——最快发现问题，最低实现成本。随着系统成熟，逐步向更高 Level 演进。

模型部署策略

业务需求分析框架

将大模型应用发布至生产环境，第一步不是技术选型，而是需求分析：

┌────────────────────────────────────────────┐
│           业务需求分析矩阵                    │
├────────────────────────────────────────────┤
│                                            │
│  功能性需求 (做什么):                         │
│  ├─ 自然语言处理 → 通用 LLM                   │
│  ├─ 代码生成 → 代码优化的 LLM                 │
│  ├─ 数学推理 → 数学微调的 LLM                 │
│  ├─ 视觉理解 → 多模态模型                     │
│  └─ 语音处理 → 语音模型                       │
│                                            │
│  非功能性需求 (如何做):                        │
│  ├─ 性能：TTFT < 500ms, TPOT < 50ms         │
│  ├─ 成本：单次调用 < $0.01                   │
│  ├─ 稳定性：99.9% 可用性                     │
│  ├─ 安全：内容过滤、隐私保护                  │
│  └─ 合规：行业监管要求                        │
│                                            │
└────────────────────────────────────────────┘

术语速查：TTFT（Time To First Token，首 token 延迟）——从发出请求到模型吐出第一个 token 的等待时间，决定用户感知的”响应快慢”；TPOT（Time Per Output Token，每 token 生成时间）——之后每多生成一个 token 的耗时，决定长输出的”流速”。两者常作为延迟 SLO 的核心指标，后文「部署前检查清单」会再次用到。

模型选型策略

并非所有场景都需要最大的模型。模型选型遵循”最小可用”原则：

任务复杂度
    ↑
    │  ┌──────────────────────────┐
    │  │ 大模型 (GPT-4, Claude)     │
    │  │ - 复杂推理                │
    │  │ - 多步规划                │
    │  │ - 创意生成                │
    │  └──────────────────────────┘
    │  ┌──────────────────────────┐
    │  │ 中模型 (GPT-4o-mini)       │
    │  │ - 意图识别                │
    │  │ - 结构化提取               │
    │  │ - 摘要生成                │
    │  └──────────────────────────┘
    │  ┌──────────────────────────┐
    │  │ 小模型 / 蒸馏模型          │
    │  │ - 文本分类                │
    │  │ - 关键词匹配               │
    │  │ - 格式校验                │
    │  └──────────────────────────┘
    └─────────────────────────────────→ 调用频率

蒸馏：让小模型掌握专业能力

本节为概览；蒸馏的完整流水线、代码与时机判断详见后文「模型蒸馏：教小模型掌握领域专业知识」一节。

蒸馏的核心思想：把大模型的判断能力”复制”给小模型。

教师模型 (GPT-4)          学生模型 (0.6B 参数)
      │                         │
      │  生成标注数据              │
      ├─────────────────────────→│
      │  "请理解请求意图"          │  学习教师的行为模式
      │  [输入→输出 对]            │
      │                         │
      │  效果：小模型在特定任务上   │
      │  接近教师模型水平          │

蒸馏 vs 微调：

蒸馏三条路径：

推理优化

性能优化框架

将 LLM 推理优化分为四个方向：

更快地处理请求

• 模型小型化：选择参数量更小的模型变体
• 量化：INT4/INT8/FP16 量化减少计算资源需求
• 剪枝：移除冗余权重，降低模型复杂度
• 知识蒸馏：用大模型数据训练小模型

量化精度对比:
  FP32 (全精度)   → 基准性能，最高计算开销
  FP16 (半精度)   → ~2x 加速，精度几乎无损
  INT8            → ~4x 加速，小幅精度损失
  INT4            → ~8x 加速，需要仔细评估精度影响

减少处理请求数

• 上下文缓存（Context Cache）：缓存多轮对话的公共前缀，减少重复计算
• 批处理（Batching）：合并多个请求为一个批次，提高硬件利用率
• 结果缓存：高频相同问题的结果直接返回缓存

# 上下文缓存的典型应用
# 多轮对话中，System Prompt + 历史知识文档是公共前缀
# 首轮：全部计算（完整价格）
# 后续轮次：缓存命中部分按 20% 价格计费

减少 Token 输入输出

• 输入端：精简输入，去除冗余信息，对长文档先生成摘要
• 输出端：通过 Prompt 引导简洁回答，设置合理的 max_tokens

max_tokens 的设计哲学：它是一个安全阀而非内容控制手段。语义完整的简短回复应使用 Prompt 引导；max_tokens 更适合作为成本控制的最后防线。

并行化处理

大模型推理本质是大规模矩阵运算。理解 CPU 和 GPU 的差异：

GPU 并行化策略：

• 数据并行：数据分片分配到多个 GPU
• 模型并行：模型不同层分布到不同设备
• 流水线并行：将计算过程划分阶段依次执行

不要默认依赖大模型

许多场景下，更简单的方法反而更高效：

安全护栏

大模型面临的安全威胁

大模型应用面临多层次的安全威胁，需要体系化的防御策略：

防御策略矩阵

安全护栏的工程实现

class SafetyGuard:
    """多层安全护栏实现"""

    def __init__(self):
        self.blocked_keywords = set()    # 自定义敏感词
        self.rate_limits = {}            # 限流记录
        self.max_tool_calls = 10         # Agent 最大工具调用次数
        self.dangerous_commands = {      # 危险命令黑名单
            "rm -rf", "DROP TABLE", "DELETE FROM",
            "os.system", "subprocess.call", "eval("
        }

    def check_input(self, user_input: str) -> tuple[bool, str]:
        """输入安全检查"""
        # 1. 敏感词检测
        for keyword in self.blocked_keywords:
            if keyword in user_input.lower():
                return False, f"输入包含敏感词: {keyword}"

        # 2. 指令注入检测
        for dangerous in self.dangerous_commands:
            if dangerous.lower() in user_input.lower():
                return False, f"检测到潜在危险指令: {dangerous}"

        return True, "通过"

    def check_tool_call(self, tool_name: str, args: dict) -> tuple[bool, str]:
        """工具调用前审计"""
        # 1. 检查调用频率
        if self.rate_limits.get(tool_name, 0) >= self.max_tool_calls:
            return False, f"工具 {tool_name} 调用次数超限"

        # 2. 检查参数安全性
        args_str = json.dumps(args).lower()
        for dangerous in self.dangerous_commands:
            if dangerous.lower() in args_str:
                return False, f"工具参数包含危险指令: {dangerous}"

        self.rate_limits[tool_name] = self.rate_limits.get(tool_name, 0) + 1
        return True, "通过"

    def check_output(self, output: str) -> tuple[bool, str]:
        """输出内容审查"""
        # 检测是否包含不应输出的敏感信息模式
        sensitive_patterns = [
            r'\b\d{17}[\dXx]\b',           # 身份证号
            r'\b1[3-9]\d{9}\b',            # 手机号
            r'[Pp]assword\s*[:=]\s*\S+',  # 密码模式
        ]

        for pattern in sensitive_patterns:
            if re.search(pattern, output):
                return False, f"输出可能包含敏感信息: {pattern}"

        return True, "通过"

熔断机制

为 Agent 的每次任务设置明确的资源上限：

class CircuitBreaker:
    """Agent 熔断器：防止失控的循环造成巨大损失"""

    def __init__(self,
                 max_api_calls: int = 10,      # 最多 API 调用次数
                 max_wall_time: int = 300,      # 最长执行时间（秒）
                 max_cost: float = 0.50):       # 最大成本（美元）
        self.max_api_calls = max_api_calls
        self.max_wall_time = max_wall_time
        self.max_cost = max_cost
        self.reset()

    def reset(self):
        self.api_calls = 0
        self.start_time = time.time()
        self.total_cost = 0.0

    def check(self) -> tuple[bool, str]:
        """检查是否应该熔断"""
        self.api_calls += 1
        elapsed = time.time() - self.start_time

        if self.api_calls > self.max_api_calls:
            return False, f"API 调用次数超限 ({self.api_calls}/{self.max_api_calls})"
        if elapsed > self.max_wall_time:
            return False, f"执行时间超限 ({elapsed:.0f}s/{self.max_wall_time}s)"
        if self.total_cost > self.max_cost:
            return False, f"成本超限 (${self.total_cost:.2f}/${self.max_cost:.2f})"

        return True, "正常"

Harness Engineering

生产环境的整体蓝图

Harness Engineering 是确保 Agent 系统从开发到生产稳定运行的全套工程实践。它包括：

可观测性

使用 OpenTelemetry 标准建立三类数据采集：

• Metrics（指标）：Token 消耗、延迟分布、错误率
• Traces（链路追踪）：一次请求经过的每个环节及耗时
• Logs（日志）：每个环节的输入输出、错误堆栈、审计信息

# 使用 OpenTelemetry 检测 Agent 调用
from opentelemetry import trace
from opentelemetry.instrumentation.openai import OpenAIInstrumentor

# 自动检测 OpenAI API 调用
OpenAIInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("agent_task")
def run_agent_task(task: str):
    # span 自动记录耗时和上下文
    with tracer.start_as_current_span("llm_call") as span:
        span.set_attribute("task", task)
        result = agent.run(task)
        span.set_attribute("tokens_used", result.usage.total_tokens)
        return result

部署前检查清单

将 Agent 推向生产前，完成以下检查：

☐ SLO 定义
  ├─ TTFT (首 Token 延迟) 目标: _____ ms
  ├─ TPOT (每 Token 生成时间) 目标: _____ ms
  └─ 可用性目标: _____%

☐ 成本控制
  ├─ 单次调用预算上限: $_____
  ├─ 日 Token 消耗上限: _____ tokens
  └─ 告警阈值已配置

☐ 安全防护
  ├─ 输入安全检查已启用
  ├─ 输出内容审查已启用
  ├─ Agent 行为熔断器已配置
  └─ 安全监控告警已配置

☐ 容灾方案
  ├─ 模型降级路径已定义
  ├─ 关键路径兜底逻辑已验证
  └─ 故障恢复演练已完成

☐ 评测基线
  ├─ 端到端评测得分: _____
  ├─ 组件级评测得分: _____
  └─ 回归测试流水线已集成

☐ 可观测性
  ├─ OpenTelemetry 已接入
  ├─ 关键指标 Dashboard 已建立
  └─ 告警规则已配置并验证

渐进式上线策略

不要一次性全量上线。采用渐进式策略：

Phase 1: 内部测试 (1-2 周)
  └─ 团队成员使用，收集初始反馈

Phase 2: 小范围灰度 (5% 用户)
  └─ 对比新旧系统的评测指标

Phase 3: 扩大范围 (25% → 50% → 100%)
  └─ 每个阶段停留 3-5 天观察指标

Phase 4: 全量上线
  └─ 保持监控，建立持续优化循环

生产运维最佳实践

评测基线管理：

• 将当前生产版本设为基线，任何新版本必须超越基线
• 定期（每周）用最新数据重新测试基线和候选版本
• 在部署流水线中集成基线检查，未达标版本自动拦截

分层降级策略：

1. 主模型不可用 → 切换到备选模型
2. 备选模型也不可用 → 使用缓存的常见回答
3. 缓存未命中 → 返回预设的降级回复模板

成本治理：

• 按模型、按用户、按任务类型维度分析 Token 消耗
• 识别异常成本峰值并设置告警
• 定期审查：是否有不必要的长上下文、冗余的 System Prompt

模型蒸馏：教小模型掌握领域专业知识

为什么蒸馏很重要

大模型（GPT-4、Claude、Qwen-72B）质量出色，但推理成本高、延迟大。对于每分钟处理数千请求的生产系统，token 成本可能非常巨大。模型蒸馏提供了一个实用的解决方案：使用大模型的输出作为训练数据，教一个小模型（7B-14B）以极低的成本复制相同的行为。

教师模型 (GPT-4, 175B 参数)
  │
  ├── 为领域任务生成高质量回答
  │
  ▼
训练数据 (输入 → 教师输出 对)
  │
  ├── 微调学生模型
  │
  ▼
学生模型 (7B-14B 参数)
  │
  ├── 相同质量，成本降低 10-50 倍
  └── 延迟降低 5-10 倍

蒸馏流水线

第 1 步：定义任务范围

蒸馏在任务定义明确且重复性高时效果最好。确定需要小模型执行的具体领域：

# 示例：客户支持意图分类
task_definitions = [
    {
        "name": "intent_classification",
        "input_schema": {"user_message": "string", "context": "string"},
        "output_schema": {"intent": "string", "confidence": "float", "reasoning": "string"},
    },
    {
        "name": "response_generation",
        "input_schema": {"intent": "string", "knowledge": "string", "tone": "string"},
        "output_schema": {"response": "string", "sources": ["string"]},
    },
]

第 2 步：用教师模型生成训练数据

使用大模型为你的领域生成高质量样本：

from openai import OpenAI

teacher = OpenAI(api_key="...")  # GPT-4 或类似模型
student = OpenAI(base_url="http://localhost:8000/v1")  # 你的小模型

def generate_training_examples(task_def: dict, n_examples: int = 1000) -> list:
    examples = []
    for i in range(n_examples):
        # 生成多样化的输入
        input_prompt = f"为任务 '{task_def['name']}' 生成一个真实的输入。变化复杂度和边界情况。"
        input_resp = teacher.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": input_prompt}],
        )
        user_input = input_resp.choices[0].message.content

        # 生成教师输出
        teacher_resp = teacher.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"你是 {task_def['name']} 的专家。遵循此输出 schema：{task_def['output_schema']}"},
                {"role": "user", "content": user_input},
            ],
        )
        teacher_output = teacher_resp.choices[0].message.content

        examples.append({"input": user_input, "output": teacher_output})

    return examples

training_data = generate_training_examples(task_definitions[0], n_examples=2000)

第 3 步：微调学生模型

使用 LoRA（Low-Rank Adaptation，低秩适配） 或全量微调，基于教师的输出来训练小模型。全量微调会更新模型所有参数，效果好但显存开销大；LoRA 则冻结原始权重不动，只在每层旁边”挂”两个很小的矩阵 A、B（其乘积 B·A 近似权重增量），只训练这两个小矩阵。因为参数量骤降（见下方输出 trainable%: 0.0622，即只训 0.06% 的参数），单卡就能跑，是现在微调大模型的主流做法。

# 使用 Hugging Face Transformers + PEFT 进行 LoRA 微调
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig, get_peft_model

model_name = "meta-llama/Llama-2-7b-hf"
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# 配置 LoRA
lora_config = LoraConfig(
    r=16,              # 秩：A/B 的"瓶颈"宽度，越大表达力越强但参数越多；16 是常用起点
    lora_alpha=32,     # 缩放系数：最终权重增量为 (alpha/r)·B·A，通常设为 r 的 2 倍
    target_modules=["q_proj", "v_proj"],  # 在哪些线性层挂 LoRA；q_proj/v_proj 即第 1 章
                                          # Attention 里的 Q、V 投影矩阵，是性价比最高的挂载点
    lora_dropout=0.05, # 对 LoRA 权重做 dropout，防过拟合
    bias="none",       # 不训练偏置项，进一步省参数
    task_type="CAUSAL_LM",
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出：trainable params: 4,194,304 || all params: 6,742,609,920 || trainable%: 0.0622
#       ↑ 67 亿参数里只训 419 万，所以单卡可跑

# 在蒸馏数据上训练
# ...（标准训练循环）

第 4 步：评估与迭代

在留出的测试集上，将学生模型的输出与教师的输出进行对比：

def evaluate_student_vs_teacher(test_set: list, teacher_client, student_client) -> dict:
    results = {"exact_match": 0, "semantic_similarity": 0, "total": len(test_set)}

    for example in test_set:
        student_resp = student_client.chat.completions.create(
            model="student-7b",
            messages=[{"role": "user", "content": example["input"]}],
        )
        student_output = student_resp.choices[0].message.content
        teacher_output = example["output"]

        # 精确匹配检查
        if student_output.strip() == teacher_output.strip():
            results["exact_match"] += 1

        # 语义相似度（使用 embedding）
        # ...（计算输出之间的余弦相似度）

    results["exact_match_rate"] = results["exact_match"] / results["total"]
    return results

# 目标：与教师的语义相似度 >85%

何时应该蒸馏

生产最佳实践：监控、灰度发布、A/B 测试

可观测性技术栈

可观测性的原理与 OpenTelemetry 实践详见前文「Harness Engineering」一节的「可观测性」小节；本节聚焦生产侧的监控层落地。

生产环境的 Agent 系统需要全面的可观测性。建立以下监控层：

1. 应用层指标

import time
from prometheus_client import Counter, Histogram, Gauge

# 请求指标
request_counter = Counter('agent_requests_total', '总请求数', ['model', 'intent', 'status'])
request_latency = Histogram('agent_request_duration_seconds', '请求延迟', ['model', 'intent'])
active_sessions = Gauge('agent_active_sessions', '活跃会话数')

# Token 使用量指标
token_usage = Counter('agent_tokens_total', 'Token 使用量', ['model', 'type'])  # type: input/output

# 质量指标
hallucination_rate = Gauge('agent_hallucination_rate', '预估幻觉率')
user_satisfaction = Histogram('agent_user_satisfaction', '用户满意度评分', buckets=[1, 2, 3, 4, 5])

def track_request(model: str, intent: str, status: str, latency: float, tokens_in: int, tokens_out: int):
    request_counter.labels(model=model, intent=intent, status=status).inc()
    request_latency.labels(model=model, intent=intent).observe(latency)
    token_usage.labels(model=model, type='input').inc(tokens_in)
    token_usage.labels(model=model, type='output').inc(tokens_out)

2. 分布式链路追踪

使用 OpenTelemetry 追踪请求在 Agent 流水线中的完整路径：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

def process_user_request(user_input: str):
    with tracer.start_as_current_span("process_request") as span:
        span.set_attribute("user.input_length", len(user_input))

        with tracer.start_as_current_span("retrieve_context") as retrieve_span:
            context = retrieve_relevant_docs(user_input)
            retrieve_span.set_attribute("docs.retrieved", len(context))

        with tracer.start_as_current_span("llm_generate") as llm_span:
            llm_span.set_attribute("model", "gpt-4")
            response = call_llm(user_input, context)
            llm_span.set_attribute("tokens.used", response.usage.total_tokens)

        return response