完整解决方案

工具调用序列记录与 Jaccard 计算

import json
from collections import OrderedDict

class ToolSequenceTracker:
    def __init__(self):
        self.session_sequences = []
        self.current_sequence = []

    def record_tool_call(self, tool_name, call_id):
        self.current_sequence.append({
            "tool": tool_name,
            "id": call_id
        })

    def finalize_session(self, session_id):
        self.session_sequences.append({
            "session_id": session_id,
            "sequence": list(self.current_sequence)
        })
        self.current_sequence = []

    def jaccard_distance(self, seq_a, seq_b):
        tools_a = set(item["tool"] for item in seq_a)
        tools_b = set(item["tool"] for item in seq_b)
        if not tools_a and not tools_b:
            return 0.0
        intersection = tools_a & tools_b
        union = tools_a | tools_b
        return 1.0 - len(intersection) / len(union)

    def check_boundary_drift(self, threshold=0.4):
        if len(self.session_sequences) < 2:
            return None
        prev = self.session_sequences[-2]["sequence"]
        curr = self.session_sequences[-1]["sequence"]
        distance = self.jaccard_distance(prev, curr)
        return {
            "distance": distance,
            "drift_detected": distance > threshold,
            "threshold": threshold
        }

词表覆盖率跟踪

import re
from collections import Counter

class LexiconDecayMonitor:
    def __init__(self, domain_terms=None):
        self.baseline_terms = set(domain_terms or [])
        self.session_coverage = []

    def update_baseline(self, text):
        words = re.findall(r'\b[a-zA-Z]{4,}\b', text.lower())
        self.baseline_terms.update(words)

    def compute_coverage(self, text):
        if not self.baseline_terms:
            return 1.0
        words = set(re.findall(r'\b[a-zA-Z]{4,}\b', text.lower()))
        hits = self.baseline_terms & words
        return len(hits) / len(self.baseline_terms)

    def check_decay(self, threshold=0.6):
        if not self.session_coverage:
            return None
        latest = self.session_coverage[-1]
        return {
            "coverage": round(latest, 3),
            "decay_detected": latest < threshold,
            "threshold": threshold
        }

语义漂移检测（基于嵌入相似度）

import numpy as np
from typing import List

class SemanticDriftDetector:
    def __init__(self, embed_fn):
        self.embed_fn = embed_fn
        self.session_embeddings = []

    def record_session(self, keywords: List[str]):
        if not keywords:
            return
        vector = self.embed_fn(keywords)
        self.session_embeddings.append(vector)

    def cosine_similarity(self, vec_a, vec_b):
        dot = np.dot(vec_a, vec_b)
        norm_a = np.linalg.norm(vec_a)
        norm_b = np.linalg.norm(vec_b)
        if norm_a == 0 or norm_b == 0:
            return 1.0
        return dot / (norm_a * norm_b)

    def check_drift(self, threshold=0.7):
        if len(self.session_embeddings) < 2:
            return None
        prev = self.session_embeddings[-2]
        curr = self.session_embeddings[-1]
        similarity = self.cosine_similarity(prev, curr)
        return {
            "similarity": round(similarity, 3),
            "drift_detected": similarity < threshold,
            "threshold": threshold
        }

在 CrewAI 中集成埋点

from crewai import Agent, Task, Crew

tracker = ToolSequenceTracker()
lexicon_monitor = LexiconDecayMonitor(domain_terms=["vector", "embedding", "context"])
drift_detector = SemanticDriftDetector(embed_fn=your_embedding_function)

session_counter = {"count": 0}

def before_agent_call(agent, task, context):
    session_counter["count"] += 1
    return {}

def after_agent_call(agent, task, output):
    output_text = str(output)
    lexicon_monitor.compute_coverage(output_text)
    lexicon_monitor.session_coverage.append(
        lexicon_monitor.compute_coverage(output_text)
    )
    drift_detector.record_session(
        lexicon_monitor.baseline_terms & set(output_text.split())
    )
    return {"status": "completed"}

agent = Agent(
    role="Research Analyst",
    goal="Analyze data and produce reports",
    memory=True,
    callbacks=[before_agent_call, after_agent_call]
)

参数与对比表格