2026年4月刘小排深海圈AI编程第三期核心干货：智能体协同与RAG生产级实践

# mcp_server.py
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.types as types

server = Server("log-analyzer")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="fetch_recent_logs",
            description="获取最近N分钟的错误日志",
            inputSchema={
                "type": "object",
                "properties": {"minutes": {"type": "integer", "default": 60}},
                "required": ["minutes"]
            }
        ),
        types.Tool(
            name="find_top_errors",
            description="统计错误类型出现次数",
            inputSchema={
                "type": "object",
                "properties": {"logs": {"type": "array", "items": {"type": "string"}}}
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "fetch_recent_logs":
        # 实际从ELK或S3拉取
        logs = ["TimeoutError", "ConnectionRefused", "TimeoutError", "NullPointer"]
        return [types.TextContent(type="text", text="\n".join(logs))]
    elif name == "find_top_errors":
        from collections import Counter
        logs = arguments.get("logs", [])
        counter = Counter(logs)
        top = counter.most_common(3)
        return [types.TextContent(type="text", text=str(top))]
    raise ValueError(f"Unknown tool: {name}")

from mcp.client import Client
import asyncio

async def main():
    async with Client("localhost", 8000) as client:
        # 主智能体提示词
        prompt = """
        你是运维专家。请按以下步骤分析：
        1. 获取最近30分钟错误日志
        2. 统计错误类型TOP3
        3. 针对最频繁的错误给出修复建议
        最后输出结构化报告。
        """
        result = await client.chat_with_tools(prompt)
        print(result)

asyncio.run(main())

from sentence_transformers import CrossEncoder
import numpy as np

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

def rerank_documents(query, docs, top_k=5):
    pairs = [[query, doc.page_content] for doc in docs]
    scores = reranker.predict(pairs)
    sorted_indices = np.argsort(scores)[::-1][:top_k]
    return [docs[i] for i in sorted_indices]

# 在RAG pipeline中嵌入
retrieved = vector_store.similarity_search(query, k=20)  # 先粗召
reranked = rerank_documents(query, retrieved, top_k=5)   # 再精排
answer = llm.generate(query, reranked)

def compress_docs(docs, max_tokens=1000):
    combined = "\n".join([d.page_content for d in docs])
    compress_prompt = f"将以下内容压缩为不超过{max_tokens}个字符的关键信息摘要：\n{combined}"
    compressed = llm.invoke(compress_prompt)
    return compressed

from opentelemetry import trace
from prometheus_client import Counter, Histogram
import functools
import time

ai_call_counter = Counter('ai_calls_total', 'Total AI calls', ['model'])
ai_latency_hist = Histogram('ai_latency_seconds', 'AI call latency')

def trace_ai(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        tracer = trace.get_tracer(__name__)
        with tracer.start_as_current_span(func.__name__):
            start = time.time()
            result = func(*args, **kwargs)
            latency = time.time() - start
            ai_call_counter.labels(model=kwargs.get('model', 'default')).inc()
            ai_latency_hist.observe(latency)
            # 记录输入输出长度到span属性
            span = trace.get_current_span()
            span.set_attribute("input_length", len(str(args)))
            span.set_attribute("output_length", len(str(result)))
            return result
    return wrapper

@trace_ai
def my_ai_function(prompt, model="gpt-4"):
    # 实际调用LLM
    return llm.invoke(prompt)