Thought: 用戶想排查 OOM 問題，我需要先查看日志
Action: 調(diào)用 SLS 日志查詢工具，搜索最近 1 小時(shí)的 OOM 日志
Observation: 找到 3 條 OOM 記錄，堆棧指向 UserService.batchQuery()
Thought: 定位到了具體方法，我需要看看這個(gè)方法的代碼
Action: 讀取 UserService.java 文件
Observation: batchQuery() 方法中有一個(gè)未分頁的全量查詢
Thought: 找到根因了——全量查詢導(dǎo)致內(nèi)存溢出
Action: 輸出分析報(bào)告和修復(fù)建議

Plan:
  Step 1: 查詢 OOM 日志
  Step 2: 分析堆棧定位代碼
  Step 3: 讀取相關(guān)源碼
  Step 4: 查詢最近代碼變更
  Step 5: 生成分析報(bào)告

Execute:
  [按計(jì)劃逐步執(zhí)行，每步完成后檢查是否需要調(diào)整計(jì)劃]

class Agent:
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools          # 可用工具列表
        self.memory = memory        # 記憶模塊

    def run(self, user_task: str) -> str:
        """Agent 的核心執(zhí)行循環(huán)"""
        self.memory.add("user", user_task)

        max_iterations = 10  # 防止無限循環(huán)
        for i in range(max_iterations):
            # 1. LLM 根據(jù)當(dāng)前上下文進(jìn)行推理
            response = self.llm.chat(
                messages=self.memory.get_messages(),
                tools=self.tools.get_schemas()  # 告訴 LLM 有哪些工具可用
            )

            # 2. 如果 LLM 決定調(diào)用工具
            if response.has_tool_call():
                tool_name = response.tool_call.name
                tool_args = response.tool_call.arguments

                # 3. 執(zhí)行工具調(diào)用
                result = self.tools.execute(tool_name, tool_args)

                # 4. 將工具結(jié)果加入記憶，供下一輪推理使用
                self.memory.add("tool_result", result)
                continue

            # 5. 如果 LLM 認(rèn)為任務(wù)已完成，直接返回結(jié)果
            if response.is_final_answer():
                return response.content

        return "達(dá)到最大執(zhí)行次數(shù)，任務(wù)未能完成"

# 常見的切分策略
class ChunkingStrategy:
    """文檔切分策略"""

    @staticmethod
    def fixed_size(text: str, chunk_size=512, overlap=50) -> list:
        """固定大小切分 —— 簡單但可能切斷語義"""
        chunks = []
        for i in range(0, len(text), chunk_size - overlap):
            chunks.append(text[i:i + chunk_size])
        return chunks

    @staticmethod
    def semantic_split(text: str) -> list:
        """語義切分 —— 按段落、章節(jié)等自然邊界切分"""
        # 優(yōu)先按標(biāo)題、段落切分，保持語義完整性
        sections = split_by_headers(text)
        chunks = []
        for section in sections:
            if len(section) > MAX_CHUNK_SIZE:
                # 超長段落再按句子切分
                chunks.extend(split_by_sentences(section))
            else:
                chunks.append(section)
        return chunks

from openai import OpenAI

client = OpenAI()

def embed_chunks(chunks: list[str]) -> list[list[float]]:
    """將文本塊批量轉(zhuǎn)換為向量"""
    response = client.embeddings.create(
        model="text-embedding-3-small",  # OpenAI 的 Embedding 模型
        input=chunks
    )
    # 返回每個(gè) chunk 對應(yīng)的向量（1536 維）
    return [item.embedding for item in response.data]

import chromadb

# 初始化向量數(shù)據(jù)庫（以 ChromaDB 為例）
client = chromadb.Client()
collection = client.create_collection("company_docs")

# 存入文檔塊及其向量
collection.add(
    ids=[f"chunk_{i}" for i in range(len(chunks))],
    documents=chunks,              # 原始文本
    embeddings=embed_chunks(chunks),  # 向量
    metadatas=[{                   # 元數(shù)據(jù)（用于過濾）
        "source": "employee_handbook.pdf",
        "chapter": "leave_policy",
        "updated_at": "2025-03-01"
    } for _ in chunks]
)

def retrieve(query: str, top_k=5) -> list[str]:
    """檢索與查詢最相關(guān)的文檔塊"""
    query_embedding = embed_chunks([query])[0]

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        where={"updated_at": {"$gte": "2025-01-01"}}  # 可選：元數(shù)據(jù)過濾
    )
    return results["documents"][0]

def generate_answer(query: str, retrieved_docs: list[str]) -> str:
    """基于檢索結(jié)果生成回答"""
    context = "\n\n---\n\n".join(retrieved_docs)

    prompt = f"""基于以下參考資料回答用戶的問題。
如果參考資料中沒有相關(guān)信息，請明確說明"根據(jù)現(xiàn)有資料無法回答"。
請?jiān)诨卮鹬袠?biāo)注信息來源。

## 參考資料
{context}

## 用戶問題
{query}

## 回答"""

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

def rewrite_query(original_query: str) -> list[str]:
    """將原始查詢改寫為多個(gè)更精確的檢索查詢"""
    prompt = f"""請將以下查詢改寫為 3 個(gè)更具體的搜索查詢，以提高檢索效果：

原始查詢：{original_query}

改寫后的查詢（每行一個(gè)）："""

    # 例如 "請假怎么操作" 會被改寫為：
    # 1. "員工請假審批流程步驟"
    # 2. "年假事假病假申請方式"
    # 3. "OA系統(tǒng)請假操作指南"
    ...

def hybrid_search(query: str, top_k=5) -> list[str]:
    """混合檢索：向量相似度 + BM25 關(guān)鍵詞匹配"""
    # 向量檢索結(jié)果
    vector_results = vector_search(query, top_k=top_k)

    # 關(guān)鍵詞檢索結(jié)果（BM25 算法）
    keyword_results = bm25_search(query, top_k=top_k)

    # 用 RRF（Reciprocal Rank Fusion）融合兩路結(jié)果
    merged = reciprocal_rank_fusion(vector_results, keyword_results)
    return merged[:top_k]

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, documents: list[str], top_k=3) -> list[str]:
    """使用 Cross-Encoder 對檢索結(jié)果重排序"""
    pairs = [(query, doc) for doc in documents]
    scores = reranker.predict(pairs)

    # 按相關(guān)性分?jǐn)?shù)排序，只保留 top_k
    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked[:top_k] if score > 0.5]  # 過濾低分文檔

# 一個(gè) Skill 的結(jié)構(gòu)描述（以代碼審查 Skill 為例）
name: "code-review"
description: "對代碼變更進(jìn)行安全性、性能、可維護(hù)性審查"
version: "1.2.0"

# 觸發(fā)條件：什么時(shí)候激活這個(gè) Skill
triggers:
  - "review this code"
  - "代碼審查"
  - "幫我 review"

# 輸入?yún)?shù)定義
inputs:
  - name: "code_diff"
    type: "string"
    required: true
    description: "需要審查的代碼變更（diff 格式）"
  - name: "language"
    type: "string"
    required: false
    default: "auto-detect"
  - name: "focus_areas"
    type: "list"
    required: false
    default: ["security", "performance", "maintainability"]

# Prompt 模板（核心）
prompt_template: |
  你是一位資深的 {{language}} 代碼審查專家。
  請對以下代碼變更進(jìn)行審查，重點(diǎn)關(guān)注：{{focus_areas}}

  ## 審查標(biāo)準(zhǔn)
  1. 安全性：是否存在注入、XSS、敏感信息泄露等風(fēng)險(xiǎn)
  2. 性能：是否有 N+1 查詢、內(nèi)存泄漏、不必要的循環(huán)
  3. 可維護(hù)性：命名是否清晰、是否符合項(xiàng)目規(guī)范

  ## 代碼變更
  {{code_diff}}

  ## 輸出格式
  按嚴(yán)重程度（Critical/Warning/Info）分類列出問題，
  每個(gè)問題給出具體的行號、問題描述和修復(fù)建議。

# 綁定的工具
tools:
  - "file_reader"      # 讀取完整文件上下文
  - "git_log"          # 查看變更歷史
  - "grep"             # 搜索相關(guān)代碼

# 輸出格式定義
output_format:
  type: "structured"
  schema:
    issues: list[{severity, line, description, suggestion}]
    summary: string
    approval: boolean

class SkillBasedAgent:
    """基于 Skill 的 Agent"""

    def __init__(self, llm):
        self.llm = llm
        self.skills = {}  # 已注冊的 Skills

    def register_skill(self, skill: Skill):
        """注冊一個(gè) Skill"""
        self.skills[skill.name] = skill

    def run(self, user_input: str) -> str:
        # 1. 意圖識別：判斷應(yīng)該使用哪個(gè) Skill
        matched_skill = self.match_skill(user_input)

        if matched_skill:
            # 2. 提取 Skill 所需的參數(shù)
            params = matched_skill.extract_params(user_input)
            # 3. 使用 Skill 的專業(yè) Prompt 和工具來執(zhí)行
            return matched_skill.execute(self.llm, params)
        else:
            # 4. 沒有匹配的 Skill，走通用對話
            return self.llm.chat(user_input)

    def match_skill(self, user_input: str) -> Skill | None:
        """匹配最適合的 Skill"""
        for skill in self.skills.values():
            if skill.can_handle(user_input):
                return skill
        return None

1. Client → Server: initialize（初始化握手）
   Client 發(fā)送自己支持的協(xié)議版本和能力

2. Server → Client: initialize response
   Server 返回自己的能力列表（支持哪些 Tools/Resources/Prompts）

3. Client → Server: tools/list（查詢可用工具）
   獲取 Server 提供的所有工具的名稱、描述、參數(shù) Schema

4. [用戶提問，LLM 決定需要調(diào)用某個(gè)工具]

5. Client → Server: tools/call（調(diào)用工具）
   {
     "method": "tools/call",
     "params": {
       "name": "query_database",
       "arguments": {
         "sql": "SELECT * FROM users WHERE status = 'active'"
       }
     }
   }

6. Server → Client: tool result（返回工具執(zhí)行結(jié)果）
   {
     "content": [
       {
         "type": "text",
         "text": "查詢到 42 條記錄..."
       }
     ]
   }

7. [LLM 基于工具返回結(jié)果繼續(xù)推理或回復(fù)用戶]

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

# 創(chuàng)建 MCP Server 實(shí)例
server = Server("weather-exchange-server")


# 定義工具列表
@server.list_tools()
async def list_tools() -> list[Tool]:
    """聲明這個(gè) Server 提供哪些工具"""
    return [
        Tool(
            name="get_weather",
            description="查詢指定城市的當(dāng)前天氣",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名稱，如 '北京'、'上海'"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="get_exchange_rate",
            description="查詢貨幣匯率",
            inputSchema={
                "type": "object",
                "properties": {
                    "from_currency": {"type": "string", "description": "源貨幣，如 USD"},
                    "to_currency": {"type": "string", "description": "目標(biāo)貨幣，如 CNY"}
                },
                "required": ["from_currency", "to_currency"]
            }
        )
    ]


# 實(shí)現(xiàn)工具的具體邏輯
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """處理工具調(diào)用請求"""
    if name == "get_weather":
        city = arguments["city"]
        # 實(shí)際項(xiàng)目中這里會調(diào)用真實(shí)的天氣 API
        weather = fetch_weather_api(city)
        return [TextContent(type="text", text=f"{city}當(dāng)前天氣：{weather}")]

    elif name == "get_exchange_rate":
        rate = fetch_exchange_rate(
            arguments["from_currency"],
            arguments["to_currency"]
        )
        return [TextContent(
            type="text",
            text=f"1 {arguments['from_currency']} = {rate} {arguments['to_currency']}"
        )]

    raise ValueError(f"未知工具: {name}")


# 啟動(dòng) Server（使用 stdio 傳輸）
async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(read, write)

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

{
  "mcpServers": {
    "weather-exchange": {
      "command": "python",
      "args": ["path/to/weather_server.py"]
    }
  }
}

用戶下達(dá)任務(wù)
    ↓
Agent（自主規(guī)劃和執(zhí)行）
    ├── 需要知識 → 調(diào)用 RAG 檢索相關(guān)文檔
    ├── 匹配到專業(yè)任務(wù) → 激活對應(yīng)的 Skill 執(zhí)行
    └── 需要調(diào)外部工具 → 通過 MCP 協(xié)議調(diào)用 MCP Server