RAG与Agent协同调优：从检索延迟到决策质量的系统性优化

from qdrant_client import QdrantClient, models
from qdrant_client.http.models import VectorParams, HnswConfigDiff, QuantizationConfig, ScalarQuantization

# 混合索引构建
client.create_collection(
    collection_name="hybrid_agent",
    vectors_config=VectorParams(size=768, distance="Cosine"),
    hnsw_config=HnswConfigDiff(
        m=32,              # 降低内存，增加ef_construct延迟
        ef_construct=100,
        on_disk=True       # 冷数据落盘
    ),
    quantization_config=ScalarQuantization(
        scalar=models.ScalarQuantization(
            type=models.ScalarType.INT8,
            quantile=0.99,
            always_ram=True
        )
    )
)

# 查询时动态调整nprobe
def adaptive_nprobe(query_vector, top_k, latency_budget_ms):
    if latency_budget_ms < 20:
        return 8
    elif latency_budget_ms < 50:
        return 32
    else:
        return 128

search_result = client.search(
    collection_name="hybrid_agent",
    query_vector=query_vec,
    limit=top_k,
    search_params=models.SearchParams(
        hnsw_ef=128,
        exact=False,
        quantization=models.QuantizationSearchParams(ignore=False)
    ),
    with_payload=True,
    # 利用client级别设置nprobe（通过grpc timeout间接控制）
)

from datasketch import MinHashLSHForest
import numpy as np

class LSHAPIIndex:
    def __init__(self, num_perm=128):
        self.forest = MinHashLSHForest(num_perm=num_perm)
        self.data = {}
    
    def add_api(self, api_name, embedding):
        # 二值化嵌入（以0为阈值）
        binary = (embedding > 0).astype(np.uint8)
        minhash = self._binary_to_minhash(binary)
        self.forest.add(api_name, minhash)
        self.data[api_name] = embedding
    
    def query(self, query_emb, top_k=5):
        binary = (query_emb > 0).astype(np.uint8)
        minhash = self._binary_to_minhash(binary)
        return self.forest.query(minhash, top_k)

def mmr_rerank(docs, query_emb, lambda_param=0.5, top_k=5):
    selected = []
    candidates = docs.copy()
    while len(selected) < top_k and candidates:
        mmr_scores = []
        for doc in candidates:
            rel = cosine_sim(query_emb, doc.embedding)
            if selected:
                max_sim = max(cosine_sim(doc.embedding, s.embedding) for s in selected)
            else:
                max_sim = 0
            mmr = lambda_param * rel - (1 - lambda_param) * max_sim
            mmr_scores.append(mmr)
        best_idx = np.argmax(mmr_scores)
        selected.append(candidates.pop(best_idx))
    return selected

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish

class EarlyStopCallback(BaseCallbackHandler):
    def __init__(self, entropy_threshold=0.85, max_steps=8):
        self.entropy_threshold = entropy_threshold
        self.max_steps = max_steps
        self.step_count = 0
        self.last_action = None
    
    def on_agent_action(self, action: AgentAction, **kwargs):
        self.step_count += 1
        # 重复动作检测
        if action.tool == self.last_action:
            raise ValueError("Loop detected, force finish")
        self.last_action = action.tool
        
        if self.step_count >= self.max_steps:
            raise ValueError("Max steps exceeded")
    
    def on_agent_finish(self, finish: AgentFinish, **kwargs):
        # 可通过finish.log解析熵值（需预先在prompt中让LLM输出confidence）
        pass

import asyncio
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=10)
async def call_tool_with_timeout(tool_func, args, timeout):
    return await asyncio.wait_for(tool_func(**args), timeout=timeout)

async def parallel_tool_invoke(tool_list):
    tasks = []
    for tool, args, timeout in tool_list:
        tasks.append(call_tool_with_timeout(tool, args, timeout))
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # 对超时或异常结果，填充默认值
    return [r if not isinstance(r, Exception) else None for r in results]

class MemoryManager:
    def __init__(self, max_tokens=4000):
        self.buffer = []
        self.summary = ""
        self.token_counter = lambda x: len(x) // 4  # 粗略估算
    
    def add(self, message):
        self.buffer.append(message)
        if self.token_counter("".join(self.buffer)) > max_tokens:
            self._summarize()
    
    def _summarize(self):
        # 调用LLM生成摘要，保留前2轮原始
        keep = self.buffer[-2:]
        to_summarize = self.buffer[:-2]
        new_summary = llm.invoke(f"Summarize: {to_summarize}")
        self.summary = new_summary
        self.buffer = [{"role": "system", "content": f"Previous: {new_summary}"}] + keep

class AdaptiveRetriever:
    def __init__(self, base_topk=10, base_lambda=0.5):
        self.topk = base_topk
        self.lambda_ = base_lambda
        self.history_scores = deque(maxlen=5)
    
    def update(self, critic_score):
        self.history_scores.append(critic_score)
        avg = sum(self.history_scores) / len(self.history_scores)
        if avg < 0.6:
            self.topk = min(20, self.topk + 5)
            self.lambda_ = min(0.9, self.lambda_ + 0.1)
        elif avg > 0.9:
            self.topk = max(3, self.topk - 2)
            self.lambda_ = max(0.2, self.lambda_ - 0.05)

import hashlib
from sentence_transformers import SentenceTransformer
from scipy.spatial.distance import cosine

class SemanticCache:
    def __init__(self, model, threshold=0.92):
        self.model = model
        self.threshold = threshold
        self.cache = {}  # emb_hash -> (response, access_time)
    
    def get(self, query):
        emb = self.model.encode(query)
        for key, (resp, _) in self.cache.items():
            if 1 - cosine(emb, key) > self.threshold:
                return resp
        return None
    
    def set(self, query, response):
        emb = self.model.encode(query)
        self.cache[tuple(emb)] = (response, time.time())

async def prefetch_on_intent(intent, top_k=5):
    tasks = []
    if "travel" in intent:
        tasks.append(async_search("北京 天气"))
        tasks.append(async_search("故宫 门票 价格"))
        tasks.append(async_search("北京 酒店 推荐"))
    return await asyncio.gather(*tasks, return_exceptions=True)