智能体构建：智能体落地80/20法则：20%模型调用，80%系统工程与策略博弈.139

import redis
import json
import openai
from functools import lru_cache

# 配置项
openai.api_key = "your-api-key"
REDIS_CONFIG = {"host": "localhost", "port": 6379, "db": 0, "decode_responses": True}
# 成本管控阈值
MAX_INPUT_TOKENS = 800
MAX_OUTPUT_TOKENS = 600
CACHE_TTL = 7200
SIMPLE_MODEL = "简单模型"
COMPLEX_MODEL = "更大的模型"

# 初始化缓存
cache_client = redis.Redis(**REDIS_CONFIG)

def estimate_tokens(text: str) -> int:
    """Token估算函数（工业级通用算法）"""
    return int(len(text) / 1.3)

def intent_classify(user_input: str) -> str:
    """意图分级：simple/complex"""
    keywords = ["分析", "诊断", "报告", "推理", "根因"]
    return "complex" if any(k in user_input for k in keywords) else "simple"

def agent_cost_control_query(user_input: str):
    # 1. 缓存拦截
    cache_key = f"agent:{hash(user_input)}"
    cached_resp = cache_client.get(cache_key)
    if cached_resp:
        return {"code": 0, "data": json.loads(cached_resp), "source": "redis_cache", "cost": 0}
    
    # 2. Token校验
    input_tokens = estimate_tokens(user_input)
    if input_tokens > MAX_INPUT_TOKENS:
        return {"code": 400, "msg": "输入内容过长，已自动裁剪管控成本"}
    
    # 3. 意图路由
    model = COMPLEX_MODEL if intent_classify(user_input) == "complex" else SIMPLE_MODEL
    
    # 4. 模型调用
    try:
        response = openai.ChatCompletion.create(
            model=model,
            messages=[{"role": "user", "content": user_input}],
            max_tokens=MAX_OUTPUT_TOKENS,
            temperature=0.2
        )
        result = response.choices[0].message.content
        total_tokens = response.usage.total_tokens
        
        # 5. 写入缓存
        cache_client.setex(cache_key, CACHE_TTL, json.dumps(result))
        return {
            "code": 0, "data": result, "source": model,
            "tokens": total_tokens, "cost": round(total_tokens * 0.0000015, 6)
        }
    except Exception as e:
        return {"code": 500, "msg": f"服务异常：{str(e)}", "source": "error"}

from langchain_community.vectorstores import FAISS
from typing import Any, List, Optional
from modelscope import snapshot_download
from sentence_transformers import SentenceTransformer
from langchain_core.embeddings import Embeddings

# 初始化向量库
embedding_model_id = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
cache_dir = "D:\\modelscope\\hub"
faiss_db_path = "D:\\medical_rag_faiss_db"  #

# 加载Embedding模型
embedding_local_path = snapshot_download(embedding_model_id, cache_dir=cache_dir, revision="master")
embedding_model = SentenceTransformer(embedding_local_path, device="cpu")

# --------------------------
# 3. 自定义LangChain Embedding适配器
# 作用：将句子转换为向量，用于语义匹配
# --------------------------
class LocalEmbedding(Embeddings):
    def __init__(self, model):
        self.model = model

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return self.model.encode(texts, convert_to_numpy=True).tolist()

    def embed_query(self, text: str) -> List[float]:
        return self.model.encode(text, convert_to_numpy=True).tolist()

embedding = LocalEmbedding(embedding_model)
vector_store = FAISS.load_local(faiss_db_path, embedding, allow_dangerous_deserialization=True)

def estimate_tokens(text: str) -> int:
    """Token估算函数（工业级通用算法）"""
    return int(len(text) / 1.3)

def precision_context_builder(user_query: str, dialogue_history: list, top_k: int = 3):
    """
    上下文精准注入核心函数
    """
    # 1. 检索高相关知识库
    relevant_docs = vector_store.similarity_search_with_score(user_query, k=top_k)
    kb_content = "\n".join([f"参考资料：{doc.page_content}" for doc, score in relevant_docs if score < 0.3])
    
    # 2. 精简对话历史（仅保留最近3轮）
    clean_history = dialogue_history[-3:] if len(dialogue_history) > 3 else dialogue_history
    history_text = "\n".join([f"{item['role']}：{item['content']}" for item in clean_history])
    
    # 3. 结构化拼接上下文
    final_context = f"""
【业务参考知识】
{kb_content if kb_content else "无额外参考资料"}

【历史对话】
{history_text if history_text else "无对话历史"}

【用户当前问题】
{user_query}

请基于以上信息，精准、简洁、无幻觉地回答问题。
"""
    return final_context, estimate_tokens(final_context)

class RobustTaskExecutor:
    def __init__(self, max_retry: int = 2, temperature: float = 0.2):
        self.max_retry = max_retry
        self.temperature = temperature

    def validate_result(self, result: str) -> bool:
        """双重校验逻辑"""
        if not result or len(result) < 5:
            return False
        # 幻觉关键词过滤
        forbidden = ["无法确认", "不存在", "虚构", "猜测"]
        return not any(k in result for k in forbidden)

    def execute_step(self, step_func, *args, **kwargs):
        """带重试+校验的步骤执行器"""
        for attempt in range(self.max_retry + 1):
            try:
                output = step_func(*args, **kwargs)
                if self.validate_result(output):
                    return {"status": "success", "data": output, "retry_times": attempt}
            except Exception as e:
                if attempt == self.max_retry:
                    return {"status": "failed", "error": str(e), "retry_times": attempt}
        return {"status": "fallback", "data": "系统繁忙，请稍后重试"}

import matplotlib.pyplot as plt
import numpy as np
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

def draw_agent_observability_dashboard():
    # 指标数据 - 9个核心指标
    # 第一排：5个百分比指标
    percent_metrics_1 = ["模型调用成功率", "缓存命中率", "步骤失败率", "幻觉检出率", "用户满意度"]
    percent_values_1 = [98.5, 72.3, 3.2, 93.5, 91.2]  # 步骤失败率用3.2%（越低越好）
    percent_colors_1 = ["#2E86AB", "#A23B72", "#F18F01", "#C73E1D", "#3A86FF"]
    
    # 第二排：绝对值指标
    abs_metrics = ["单请求Token消耗", "平均响应耗时", "单日成本统计"]
    abs_values = [1850, 1180, 126.5]  # Token数, ms, 元
    abs_colors = ["#7209B7", "#52B788", "#FB8500"]
    
    # 创建2列布局
    fig, axes = plt.subplots(1, 2, figsize=(12, 6))
    
    # 左：5个百分比指标
    x1 = range(len(percent_metrics_1))
    bars1 = axes[0].bar(x1, percent_values_1, color=percent_colors_1, edgecolor="black", linewidth=1.2)
    axes[0].set_title("质量效率指标", fontsize=14, fontweight="bold", pad=10)
    axes[0].set_ylim(0, 110)
    axes[0].set_xticks(x1)
    axes[0].set_xticklabels(percent_metrics_1, fontsize=9, rotation=20)
    axes[0].set_ylabel("百分比 (%)", fontsize=11)
    axes[0].grid(axis="y", alpha=0.3)
    
    for bar, val in zip(bars1, percent_values_1):
        axes[0].text(bar.get_x()+bar.get_width()/2., bar.get_height()+2, 
                    f"{val}%", ha="center", fontsize=9, fontweight="bold")
    
    # 右：3个绝对值指标
    x2 = range(len(abs_metrics))
    bars2 = axes[1].bar(x2, abs_values, color=abs_colors, edgecolor="black", linewidth=1.2, width=0.5)
    axes[1].set_title("性能成本指标", fontsize=14, fontweight="bold", pad=10)
    axes[1].set_ylim(0, max(abs_values) * 1.25)
    axes[1].set_xticks(x2)
    axes[1].set_xticklabels(abs_metrics, fontsize=10)
    axes[1].grid(axis="y", alpha=0.3)
    
    # 每个指标不同单位
    units = ["Tokens", "ms", "元/天"]
    for bar, val, unit in zip(bars2, abs_values, units):
        axes[1].text(bar.get_x()+bar.get_width()/2., bar.get_height() + max(abs_values)*0.05, 
                    f"{val}", ha="center", fontsize=10, fontweight="bold")
    
    fig.suptitle("智能体可观测性核心指标仪表盘", fontsize=16, fontweight="bold")
    
    plt.tight_layout()
    plt.subplots_adjust(top=0.88, wspace=0.1)
    
    plt.savefig("agent_observability.png", dpi=300)
    plt.show()
    plt.close()

draw_agent_observability_dashboard()