LangChain实战与企业级实践综合指南

import hashlib
import base64
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os
import re
from typing import Dict, Any, Optional

class LangChainSecurityManager:
    """LangChain安全管理器"""

    def __init__(self, master_key: Optional[str] = None):
        self.master_key = master_key or os.environ.get('LANGCHAIN_MASTER_KEY')
        if not self.master_key:
            raise ValueError("必须提供主密钥")

        self.cipher_suite = self._create_cipher_suite()
        self.sensitive_patterns = [
            r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',  # 信用卡号
            r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # 邮箱
            r'\b\d{11}\b',  # 手机号
        ]

    def _create_cipher_suite(self) -> Fernet:
        """创建加密套件"""
        password = self.master_key.encode()
        salt = b'langchain_salt_2024'  # 在生产环境中应使用随机salt

        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )

        key = base64.urlsafe_b64encode(kdf.derive(password))
        return Fernet(key)

    def encrypt_sensitive_data(self, data: str) -> str:
        """加密敏感数据"""
        try:
            encrypted_data = self.cipher_suite.encrypt(data.encode())
            return base64.urlsafe_b64encode(encrypted_data).decode()
        except Exception as e:
            raise ValueError(f"数据加密失败: {str(e)}")

    def decrypt_sensitive_data(self, encrypted_data: str) -> str:
        """解密敏感数据"""
        try:
            encrypted_bytes = base64.urlsafe_b64decode(encrypted_data.encode())
            decrypted_data = self.cipher_suite.decrypt(encrypted_bytes)
            return decrypted_data.decode()
        except Exception as e:
            raise ValueError(f"数据解密失败: {str(e)}")

    def sanitize_input(self, text: str) -> str:
        """清理输入中的敏感信息"""
        sanitized_text = text

        for pattern in self.sensitive_patterns:
            # 替换敏感信息为占位符
            sanitized_text = re.sub(pattern, '[REDACTED]', sanitized_text)

        return sanitized_text

    def create_secure_prompt_template(self, template: str) -> 'SecurePromptTemplate':
        """创建安全的提示模板"""
        return SecurePromptTemplate(template, self)

class SecurePromptTemplate:
    """安全的提示模板"""

    def __init__(self, template: str, security_manager: LangChainSecurityManager):
        self.template = template
        self.security_manager = security_manager

    def format(self, **kwargs) -> str:
        """格式化提示，自动清理敏感信息"""
        sanitized_kwargs = {}

        for key, value in kwargs.items():
            if isinstance(value, str):
                sanitized_kwargs[key] = self.security_manager.sanitize_input(value)
            else:
                sanitized_kwargs[key] = value

        return self.template.format(**sanitized_kwargs)

from enum import Enum
from functools import wraps
from typing import List, Dict, Any, Callable
import jwt
import time

class Permission(Enum):
    """权限枚举"""
    READ_DOCUMENTS = "read_documents"
    WRITE_DOCUMENTS = "write_documents"
    EXECUTE_TOOLS = "execute_tools"
    MANAGE_AGENTS = "manage_agents"
    ADMIN_ACCESS = "admin_access"

class Role(Enum):
    """角色枚举"""
    GUEST = "guest"
    USER = "user"
    DEVELOPER = "developer"
    ADMIN = "admin"

class AccessControlManager:
    """访问控制管理器"""

    def __init__(self, secret_key: str):
        self.secret_key = secret_key
        self.role_permissions = {
            Role.GUEST: [Permission.READ_DOCUMENTS],
            Role.USER: [Permission.READ_DOCUMENTS, Permission.EXECUTE_TOOLS],
            Role.DEVELOPER: [
                Permission.READ_DOCUMENTS,
                Permission.WRITE_DOCUMENTS,
                Permission.EXECUTE_TOOLS,
                Permission.MANAGE_AGENTS
            ],
            Role.ADMIN: [
                Permission.READ_DOCUMENTS,
                Permission.WRITE_DOCUMENTS,
                Permission.EXECUTE_TOOLS,
                Permission.MANAGE_AGENTS,
                Permission.ADMIN_ACCESS
            ]
        }

    def create_token(self, user_id: str, role: Role, expires_in: int = 3600) -> str:
        """创建JWT令牌"""
        payload = {
            'user_id': user_id,
            'role': role.value,
            'permissions': [p.value for p in self.role_permissions[role]],
            'exp': time.time() + expires_in,
            'iat': time.time()
        }

        return jwt.encode(payload, self.secret_key, algorithm='HS256')

    def verify_token(self, token: str) -> Dict[str, Any]:
        """验证JWT令牌"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            return payload
        except jwt.ExpiredSignatureError:
            raise ValueError("令牌已过期")
        except jwt.InvalidTokenError:
            raise ValueError("无效的令牌")

    def check_permission(self, token: str, required_permission: Permission) -> bool:
        """检查权限"""
        try:
            payload = self.verify_token(token)
            user_permissions = payload.get('permissions', [])
            return required_permission.value in user_permissions
        except ValueError:
            return False

    def require_permission(self, permission: Permission):
        """权限装饰器"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 从kwargs中获取token，或从请求头中获取
                token = kwargs.get('auth_token') or getattr(args[0], 'auth_token', None)

                if not token:
                    raise PermissionError("缺少认证令牌")

                if not self.check_permission(token, permission):
                    raise PermissionError(f"缺少必要权限: {permission.value}")

                return func(*args, **kwargs)
            return wrapper
        return decorator

from dataclasses import dataclass
from typing import Literal, Dict

Region = Literal["eu", "us", "apac"]

@dataclass
class DataResidencyConfig:
    tenant_id: str
    residency: Region  # 数据驻留地域
    pii_level: Literal["none", "low", "medium", "high"]
    encrypt_at_rest: bool = True
    kms_key_id: str | None = None
    retention_days: int = 180

    def storage_bucket(self) -> str:
        # 依据地域与租户路由到不同的对象存储/数据库实例
        return f"lc-{self.residency}-tenant-{self.tenant_id}"

    def should_mask_output(self) -> bool:
        return self.pii_level in ("medium", "high")

import json, time, os
from uuid import uuid4
from typing import Any, Dict, List, Optional
from langchain_core.callbacks import BaseCallbackHandler

class AuditCallbackHandler(BaseCallbackHandler):
    """结构化审计：链/LLM/工具 关键事件持久化（JSON Lines）"""

    def __init__(self, path: str = "./logs/audit.jsonl", tenant_id: str = "default", region: str = "eu"):
        os.makedirs(os.path.dirname(path), exist_ok=True)
        self.path = path
        self.tenant_id = tenant_id
        self.region = region

    def _write(self, record: Dict[str, Any]):
        record.setdefault("ts", time.time())
        record.setdefault("tenant_id", self.tenant_id)
        record.setdefault("region", self.region)
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], *, run_id, **kwargs):
        self._write({
            "event": "llm_start",
            "run_id": str(run_id),
            "model": serialized.get("id"),
            "prompt_preview": (prompts[0][:200] if prompts else ""),
        })

    def on_llm_end(self, response, *, run_id, **kwargs):
        usage = {}
        if getattr(response, "llm_output", None):
            usage = response.llm_output.get("token_usage", {})
        self._write({
            "event": "llm_end",
            "run_id": str(run_id),
            "usage": usage,
        })

from typing import Any, Dict, List, Optional, Union
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.outputs import ChatResult, ChatGeneration
import base64
import requests
from PIL import Image
import io

class MultiModalChatModel(BaseChatModel):
    """多模态聊天模型集成"""

    def __init__(
        self,
        model_name: str = "gpt-4-vision-preview",
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        max_tokens: int = 1000,
        temperature: float = 0.1,
        **kwargs
    ):
        super().__init__(**kwargs)
        self.model_name = model_name
        self.api_key = api_key
        self.base_url = base_url or "https://api.openai.com/v1"
        self.max_tokens = max_tokens
        self.temperature = temperature

        # 支持的图像格式
        self.supported_image_formats = {'.jpg', '.jpeg', '.png', '.gif', '.webp'}

        # 模态处理器注册
        self.modality_processors = {
            'text': self._process_text,
            'image': self._process_image,
            'audio': self._process_audio,
            'video': self._process_video
        }

    def _process_image(self, item: Dict[str, Any]) -> Dict[str, Any]:
        """处理图像模态"""
        image_data = item.get("image_url") or item.get("image")

        if isinstance(image_data, str):
            if image_data.startswith("http"):
                # 网络图片URL
                return {
                    "type": "image_url",
                    "image_url": {
                        "url": image_data,
                        "detail": item.get("detail", "auto")
                    }
                }
            elif image_data.startswith("data:image"):
                # Base64编码的图片
                return {
                    "type": "image_url",
                    "image_url": {
                        "url": image_data,
                        "detail": item.get("detail", "auto")
                    }
                }
            else:
                # 本地文件路径
                encoded_image = self._encode_image_file(image_data)
                return {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/jpeg;base64,{encoded_image}",
                        "detail": item.get("detail", "auto")
                    }
                }

        return {
            "type": "text",
            "text": "[无法处理的图像数据]"
        }

    def _encode_image_file(self, image_path: str) -> str:
        """编码本地图像文件为Base64"""
        try:
            with open(image_path, "rb") as image_file:
                encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
            return encoded_string
        except Exception as e:
            raise ValueError(f"无法编码图像文件 {image_path}: {str(e)}")

from typing import AsyncIterator, Callable
import asyncio
import re

SENSITIVE_PATTERNS = [
    re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),            # SSN
    re.compile(r"\b\d{16}\b"),                         # 粗略信用卡
    re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"),
]

def sanitize_chunk(text: str) -> str:
    for p in SENSITIVE_PATTERNS:
        text = p.sub("[REDACTED]", text)
    return text

async def astream_with_safety(chain, payload: dict, *, on_chunk: Callable[[str], None]) -> str:
    """边流式边过滤，返回最终完整文本"""
    full = []
    async for chunk in chain.astream(payload):
        safe = sanitize_chunk(str(chunk))
        on_chunk(safe)
        full.append(safe)
        await asyncio.sleep(0)  # 让出事件循环
    return "".join(full)

from typing import List, Dict, Any, Optional, Callable
import random
import time
import threading
from dataclasses import dataclass
from enum import Enum
import logging

class ProviderStatus(Enum):
    """Provider状态枚举"""
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"
    MAINTENANCE = "maintenance"

@dataclass
class ProviderMetrics:
    """Provider性能指标"""
    response_time: float = 0.0
    success_rate: float = 1.0
    error_count: int = 0
    total_requests: int = 0
    last_error_time: Optional[float] = None
    status: ProviderStatus = ProviderStatus.HEALTHY

class LoadBalancedChatModel:
    """负载均衡的聊天模型"""

    def __init__(
        self,
        providers: List[Dict[str, Any]],
        strategy: str = "round_robin",
        health_check_interval: int = 60,
        max_retries: int = 3,
        circuit_breaker_threshold: float = 0.5,
        **kwargs
    ):
        self.providers = {}
        self.provider_metrics = {}
        self.strategy = strategy
        self.health_check_interval = health_check_interval
        self.max_retries = max_retries
        self.circuit_breaker_threshold = circuit_breaker_threshold

        # 初始化providers
        for i, provider_config in enumerate(providers):
            provider_id = f"provider_{i}"
            self.providers[provider_id] = self._create_provider(provider_config)
            self.provider_metrics[provider_id] = ProviderMetrics()

        # 负载均衡策略
        self.current_index = 0
        self.strategy_lock = threading.Lock()

        # 健康检查
        self.health_check_thread = threading.Thread(
            target=self._health_check_loop,
            daemon=True
        )
        self.health_check_thread.start()

        self.logger = logging.getLogger(__name__)

    def _generate_with_fallback(
        self,
        messages: List[Any],
        **kwargs
    ) -> Any:
        """带故障转移的生成"""

        last_exception = None
        attempted_providers = set()

        for attempt in range(self.max_retries):
            # 选择provider
            provider_id = self._select_provider()

            if not provider_id or provider_id in attempted_providers:
                # 如果没有可用provider或已尝试过，跳出循环
                break

            attempted_providers.add(provider_id)
            provider = self.providers[provider_id]
            metrics = self.provider_metrics[provider_id]

            try:
                start_time = time.time()

                # 调用provider
                result = provider._generate(messages, **kwargs)

                # 更新成功指标
                response_time = time.time() - start_time
                self._update_success_metrics(provider_id, response_time)

                return result

            except Exception as e:
                last_exception = e

                # 更新失败指标
                self._update_failure_metrics(provider_id, e)

                self.logger.warning(
                    f"Provider {provider_id} 调用失败 (尝试 {attempt + 1}): {str(e)}"
                )

                # 如果还有重试机会，继续下一个provider
                continue

        # 所有provider都失败了
        if last_exception:
            raise last_exception
        else:
            raise RuntimeError("没有可用的provider")

import time
import asyncio
from typing import Callable, Any

class RetryPolicy:
    def __init__(self, max_attempts: int = 3, base_delay: float = 0.2, jitter: float = 0.1):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.jitter = jitter

    async def aretry(self, fn: Callable[[], Any]):
        last = None
        for i in range(self.max_attempts):
            try:
                return await fn()
            except Exception as e:
                last = e
                await asyncio.sleep(self.base_delay * (2 ** i) + self.jitter)
        raise last

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, cool_down: float = 30.0):
        self.failure_threshold = failure_threshold
        self.cool_down = cool_down
        self.failures = 0
        self.open_until = 0.0

    def allow(self) -> bool:
        return time.time() >= self.open_until

    def record_success(self):
        self.failures = 0

    def record_failure(self):
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.open_until = time.time() + self.cool_down

class RateLimiter:
    def __init__(self, qps: float = 10.0):
        self.interval = 1.0 / qps
        self.last = 0.0

    async def acquire(self):
        now = time.time()
        delta = self.interval - (now - self.last)
        if delta > 0:
            await asyncio.sleep(delta)
        self.last = time.time()

class QuotaRouter:
    """按模型/Provider 配额与成本做路由"""
    def __init__(self, providers: list[dict]):
        self.providers = providers  # [{"id": "openai:gpt-3.5", "cost": 1, "remaining": 1000}, ...]

    def select(self) -> dict:
        affordable = [p for p in self.providers if p.get("remaining", 0) > 0]
        if not affordable:
            # 全部耗尽时，选择质量更高但更贵的作为兜底
            return sorted(self.providers, key=lambda x: x["cost"])[0]
        # 选择最低成本的可用配额
        return sorted(affordable, key=lambda x: x["cost"])[0]

    def consume(self, pid: str, tokens: int):
        for p in self.providers:
            if p["id"] == pid:
                p["remaining"] = max(0, p.get("remaining", 0) - tokens)
                break

from typing import Any, Dict, List, Optional, Tuple, Union
from langchain_core.vectorstores import VectorStore
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
import numpy as np
import faiss
import pickle
import os
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

class HighPerformanceVectorStore(VectorStore):
    """高性能向量存储实现"""

    def __init__(
        self,
        embedding_function: Embeddings,
        index_factory: str = "IVF1024,Flat",
        metric_type: str = "L2",
        use_gpu: bool = False,
        cache_size: int = 10000,
        batch_size: int = 1000,
        **kwargs
    ):
        self.embedding_function = embedding_function
        self.index_factory = index_factory
        self.metric_type = metric_type
        self.use_gpu = use_gpu
        self.cache_size = cache_size
        self.batch_size = batch_size

        # 初始化FAISS索引
        self.index = None
        self.dimension = None

        # 文档存储
        self.documents = {}
        self.id_to_index = {}
        self.index_to_id = {}

        # 缓存机制
        self.query_cache = {}
        self.cache_lock = threading.RLock()

        # 性能统计
        self.stats = {
            'total_queries': 0,
            'cache_hits': 0,
            'avg_query_time': 0,
            'total_documents': 0
        }

    def similarity_search_with_score(
        self,
        query: str,
        k: int = 4,
        filter: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> List[Tuple[Document, float]]:
        """相似度搜索（带分数）"""

        start_time = time.time()

        # 检查缓存
        cache_key = self._generate_cache_key(query, k, filter)

        with self.cache_lock:
            if cache_key in self.query_cache:
                self.stats['cache_hits'] += 1
                self.stats['total_queries'] += 1
                return self.query_cache[cache_key]

        # 生成查询向量
        query_embedding = self.embedding_function.embed_query(query)
        query_vector = np.array([query_embedding], dtype=np.float32)

        # 执行搜索
        if self.index is None or self.index.ntotal == 0:
            return []

        # FAISS搜索
        scores, indices = self.index.search(query_vector, min(k, self.index.ntotal))

        # 处理结果
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx == -1:  # FAISS返回-1表示无效结果
                continue

            doc_id = self.index_to_id.get(idx)
            if doc_id and doc_id in self.documents:
                doc = self.documents[doc_id]

                # 应用过滤器
                if filter and not self._apply_filter(doc, filter):
                    continue

                # 转换分数（FAISS返回的是距离，需要转换为相似度）
                similarity_score = self._distance_to_similarity(score)
                results.append((doc, similarity_score))

        # 限制结果数量
        results = results[:k]

        # 缓存结果
        with self.cache_lock:
            if len(self.query_cache) < self.cache_size:
                self.query_cache[cache_key] = results

        # 更新统计
        query_time = time.time() - start_time
        self.stats['total_queries'] += 1

        # 更新平均查询时间
        total_queries = self.stats['total_queries']
        current_avg = self.stats['avg_query_time']
        self.stats['avg_query_time'] = (
            (current_avg * (total_queries - 1) + query_time) / total_queries
        )

        return results

    def _distance_to_similarity(self, distance: float) -> float:
        """将距离转换为相似度分数"""
        return 1.0 / (1.0 + distance)

from typing import List, Tuple

class HybridRetriever:
    """向量检索 + BM25（或关键词） 混合，并用 RRF 融合"""
    def __init__(self, vector_retriever, bm25_retriever, k: int = 8, alpha: float = 0.7, reranker=None):
        self.vec = vector_retriever
        self.bm25 = bm25_retriever
        self.k = k
        self.alpha = alpha
        self.reranker = reranker  # 可对融合后的候选做二次重排

    def _rrf(self, lists: List[List[Tuple[str, float]]]) -> List[Tuple[str, float]]:
        # Reciprocal Rank Fusion: score += 1/(rank + 60)
        scores = {}
        for results in lists:
            for rank, (doc_id, _) in enumerate(results):
                scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (rank + 60.0)
        return sorted(scores.items(), key=lambda x: x[1], reverse=True)

    def get_relevant_documents(self, query: str):
        vec_docs = self.vec.get_relevant_documents(query)
        bm25_docs = self.bm25.get_relevant_documents(query)

        fused_ids = self._rrf([self._topk_pairs(vec_docs), self._topk_pairs(bm25_docs)])

        # 恢复文档对象并截断到 k
        id_to_doc = {}
        for d in vec_docs + bm25_docs:
            id_to_doc[getattr(d, "id", id(d))] = d

        candidates = [id_to_doc[i] for i, _ in fused_ids if i in id_to_doc][: self.k]

        if self.reranker:
            candidates = self.reranker.rerank(query, candidates)[: self.k]
        return candidates

from langchain_core.documents import Document

class SemanticCache:
    def __init__(self, embeddings, vectorstore, threshold: float = 0.92):
        self.emb = embeddings
        self.vs = vectorstore
        self.threshold = threshold

    def lookup(self, query: str) -> str | None:
        results = self.vs.similarity_search_with_score(query, k=1)
        if not results:
            return None
        doc, score = results[0]
        if score >= self.threshold:  # 假设 score 越大越相似
            return doc.metadata.get("response")
        return None

    def update(self, query: str, response: str):
        doc = Document(page_content=query, metadata={"response": response})
        self.vs.add_texts([doc.page_content], metadatas=[doc.metadata])

class EnterpriseKnowledgeBase:
    """企业级知识库问答系统"""

    def __init__(self):
        # 文档处理管道
        self.document_loader = MultiSourceLoader([
            "confluence", "sharepoint", "local_files", "databases"
        ])

        # 智能分割策略
        self.text_splitter = HybridTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", "。", ".", " "]
        )

        # 向量存储
        self.vectorstore = PineconeVectorStore(
            index_name="enterprise-kb",
            namespace="production"
        )

        # 检索增强
        self.retriever = HybridRetriever(
            vector_retriever=self.vectorstore.as_retriever(),
            bm25_retriever=BM25Retriever(),
            fusion_weights=[0.7, 0.3]
        )

        # LLM 配置
        self.llm = ChatOpenAI(
            model="gpt-4",
            temperature=0.1,
            max_tokens=2000
        )

    def build_qa_chain(self):
        """构建问答链"""

        # 检索 Prompt
        retrieval_prompt = ChatPromptTemplate.from_messages([
            ("system", """你是企业内部知识库助手。基于以下检索到的文档回答问题：

文档内容：
{context}

回答要求：
1. 基于文档内容准确回答
2. 如果文档中没有相关信息，明确说明
3. 提供文档来源和页码（如有）
4. 使用专业术语，保持企业标准

问题：{question}"""),
        ])

        # 构建 RAG 链
        rag_chain = (
            {
                "context": self.retriever | self._format_docs,
                "question": RunnablePassthrough()
            }
            | retrieval_prompt
            | self.llm
            | StrOutputParser()
        )

        return rag_chain

class IntelligentCustomerService:
    """智能客服系统"""

    def __init__(self):
        # 多轮对话管理
        self.memory = ConversationSummaryBufferMemory(
            llm=ChatOpenAI(model="gpt-3.5-turbo"),
            max_token_limit=2000,
            return_messages=True
        )

        # 意图识别
        self.intent_classifier = IntentClassifier([
            "product_inquiry", "order_status", "refund_request",
            "technical_support", "complaint", "general_question"
        ])

        # 工具集
        self.tools = [
            OrderQueryTool(),
            ProductSearchTool(),
            RefundProcessTool(),
            EscalationTool()
        ]

        # Agent 配置
        self.agent = create_openai_functions_agent(
            llm=ChatOpenAI(model="gpt-4", temperature=0.1),
            tools=self.tools,
            prompt=self._create_customer_service_prompt()
        )

        self.agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            memory=self.memory,
            verbose=False,
            max_iterations=5,
            handle_parsing_errors=True
        )

    async def handle_customer_query(self, query: str, session_id: str) -> Dict[str, Any]:
        """处理客户咨询"""

        # 1. 意图识别
        intent = await self.intent_classifier.classify(query)

        # 2. 情感分析
        sentiment = await self._analyze_sentiment(query)

        # 3. Agent 处理
        response = await self.agent_executor.ainvoke({
            "input": query,
            "intent": intent,
            "sentiment": sentiment
        })

        # 4. 质量检查
        quality_score = await self._check_response_quality(query, response["output"])

        # 5. 记录日志
        await self._log_interaction(session_id, query, response, quality_score)

        return {
            "response": response["output"],
            "intent": intent,
            "sentiment": sentiment,
            "quality_score": quality_score,
            "should_escalate": quality_score < 0.7 or sentiment == "negative"
        }

# services/base.py
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from langchain_core.runnables import RunnableConfig
import logging

class BaseService(ABC):
    """基础服务类"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)
        self._initialize()

    @abstractmethod
    def _initialize(self) -> None:
        """初始化服务"""
        pass

    @abstractmethod
    def health_check(self) -> Dict[str, Any]:
        """健康检查"""
        pass

class AgentService(BaseService):
    """Agent服务 - 管理和执行Agent"""

    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        self.agents: Dict[str, AgentExecutor] = {}
        self.tools: Dict[str, BaseTool] = {}

    def execute_agent(
        self,
        agent_name: str,
        inputs: Dict[str, Any],
        config: Optional[RunnableConfig] = None
    ) -> Dict[str, Any]:
        """执行指定的Agent"""
        if agent_name not in self.agents:
            raise ValueError(f"Agent '{agent_name}' not found")

        agent = self.agents[agent_name]

        try:
            self.logger.info(f"Executing agent '{agent_name}' with inputs: {inputs}")
            result = agent.invoke(inputs, config=config)
            self.logger.info(f"Agent '{agent_name}' completed successfully")
            return result
        except Exception as e:
            self.logger.error(f"Agent '{agent_name}' execution failed: {str(e)}")
            raise

# config/manager.py
import os
import yaml
import json
from typing import Any, Dict, Optional
from pathlib import Path

class ConfigManager:
    """配置管理器 - 支持多环境、多格式配置"""

    def __init__(self, config_dir: str = "config"):
        self.config_dir = Path(config_dir)
        self.environment = os.getenv("ENVIRONMENT", "development")
        self._config_cache: Dict[str, Any] = {}
        self._load_configs()

    def _load_configs(self) -> None:
        """加载配置文件"""
        # 1. 加载基础配置
        base_config_file = self.config_dir / "base.yaml"
        if base_config_file.exists():
            with open(base_config_file, 'r') as f:
                base_config = yaml.safe_load(f)
                self._config_cache.update(base_config)

        # 2. 加载环境特定配置
        env_config_file = self.config_dir / f"{self.environment}.yaml"
        if env_config_file.exists():
            with open(env_config_file, 'r') as f:
                env_config = yaml.safe_load(f)
                self._deep_merge(self._config_cache, env_config)

        # 3. 环境变量覆盖
        self._apply_env_overrides()

    def get_llm_config(self, model_name: str = None) -> Dict[str, Any]:
        """获取LLM配置"""
        model_name = model_name or self.get("llm.default_model", "gpt-3.5-turbo")

        base_config = self.get("llm.base_config", {})
        model_config = self.get(f"llm.models.{model_name}", {})

        # 合并配置
        config = {**base_config, **model_config}

        # 添加API密钥
        if "openai" in model_name.lower():
            config["openai_api_key"] = os.getenv("OPENAI_API_KEY")
        elif "anthropic" in model_name.lower():
            config["anthropic_api_key"] = os.getenv("ANTHROPIC_API_KEY")

        return config

# optimization/llm_cache.py
import hashlib
import json
import time
from typing import Any, Dict, List, Optional, Tuple
from langchain_core.caches import BaseCache
import redis
import pickle

class MultiLevelCache(BaseCache):
    """多级缓存系统 - 内存 + Redis + 持久化"""

    def __init__(
        self,
        memory_size: int = 1000,
        redis_client: Optional[redis.Redis] = None,
        redis_ttl: int = 3600,
        persistent_cache_file: Optional[str] = None
    ):
        self.memory_cache: Dict[str, Tuple[Any, float]] = {}
        self.memory_size = memory_size
        self.redis_client = redis_client
        self.redis_ttl = redis_ttl
        self.persistent_cache_file = persistent_cache_file

    def lookup(self, prompt: str, llm_string: str) -> Optional[List[Any]]:
        """查找缓存"""
        key = self._generate_key(prompt, llm_string)

        # 1. 检查内存缓存
        if key in self.memory_cache:
            value, timestamp = self.memory_cache[key]
            # 检查是否过期（内存缓存1小时过期）
            if time.time() - timestamp < 3600:
                return value
            else:
                del self.memory_cache[key]

        # 2. 检查Redis缓存
        if self.redis_client:
            try:
                cached_data = self.redis_client.get(f"llm_cache:{key}")
                if cached_data:
                    value = pickle.loads(cached_data)
                    # 更新到内存缓存
                    self._update_memory_cache(key, value)
                    return value
            except Exception as e:
                print(f"Redis cache lookup failed: {e}")

        return None

    def update(self, prompt: str, llm_string: str, return_val: List[Any]) -> None:
        """更新缓存"""
        key = self._generate_key(prompt, llm_string)

        # 1. 更新内存缓存
        self._update_memory_cache(key, return_val)

        # 2. 更新Redis缓存
        if self.redis_client:
            try:
                cached_data = pickle.dumps(return_val)
                self.redis_client.setex(
                    f"llm_cache:{key}",
                    self.redis_ttl,
                    cached_data
                )
            except Exception as e:
                print(f"Redis cache update failed: {e}")

# optimization/batch_processor.py
import asyncio
from typing import Any, Dict, List, Optional
import time

class BatchProcessor:
    """批量处理器 - 优化LLM调用性能"""

    def __init__(
        self,
        batch_size: int = 10,
        max_wait_time: float = 1.0,
        max_concurrent: int = 5
    ):
        self.batch_size = batch_size
        self.max_wait_time = max_wait_time
        self.max_concurrent = max_concurrent
        self.pending_requests: List[Dict[str, Any]] = []
        self.request_futures: Dict[str, asyncio.Future] = {}
        self._processing = False

    async def process_request(
        self,
        llm,
        prompt: str,
        **kwargs
    ) -> Any:
        """处理单个请求（可能被批量化）"""
        request_id = f"{time.time()}_{len(self.pending_requests)}"

        # 创建Future用于返回结果
        future = asyncio.Future()
        self.request_futures[request_id] = future

        # 添加到待处理队列
        request = {
            'id': request_id,
            'llm': llm,
            'prompt': prompt,
            'kwargs': kwargs
        }
        self.pending_requests.append(request)

        # 如果达到批量大小或者是第一个请求，开始处理
        if len(self.pending_requests) >= self.batch_size or not self._processing:
            asyncio.create_task(self._process_batch())

        # 等待结果
        return await future

    async def _process_llm_group(self, llm, requests: List[Dict[str, Any]]) -> None:
        """处理同一LLM的请求组"""
        try:
            # 检查LLM是否支持批量处理
            if hasattr(llm, 'agenerate') and len(requests) > 1:
                # 批量处理
                prompts = [req['prompt'] for req in requests]

                # 合并kwargs（假设同组请求的kwargs相同）
                common_kwargs = requests[0]['kwargs']

                # 批量调用
                results = await llm.agenerate(prompts, **common_kwargs)

                # 分发结果
                for i, request in enumerate(requests):
                    future = self.request_futures.pop(request['id'])
                    if not future.done():
                        future.set_result(results.generations[i][0].text)

            else:
                # 并发单独处理
                semaphore = asyncio.Semaphore(self.max_concurrent)

                async def process_single(request):
                    async with semaphore:
                        try:
                            result = await llm.agenerate([request['prompt']], **request['kwargs'])
                            future = self.request_futures.pop(request['id'])
                            if not future.done():
                                future.set_result(result.generations[0][0].text)
                        except Exception as e:
                            future = self.request_futures.pop(request['id'])
                            if not future.done():
                                future.set_exception(e)

                tasks = [process_single(req) for req in requests]
                await asyncio.gather(*tasks, return_exceptions=True)

        except Exception as e:
            # 处理组级错误
            for request in requests:
                future = self.request_futures.pop(request['id'], None)
                if future and not future.done():
                    future.set_exception(e)

# optimization/token_optimizer.py
from typing import List, Dict, Any
import tiktoken

class TokenOptimizer:
    """Token使用优化器"""

    def __init__(self, model: str = "gpt-3.5-turbo"):
        self.model = model
        self.encoding = tiktoken.encoding_for_model(model)

        # 模型token限制
        self.token_limits = {
            "gpt-3.5-turbo": 4096,
            "gpt-3.5-turbo-16k": 16384,
            "gpt-4": 8192,
            "gpt-4-32k": 32768,
        }

        self.max_tokens = self.token_limits.get(model, 4096)

    def count_tokens(self, text: str) -> int:
        """计算文本token数量"""
        return len(self.encoding.encode(text))

    def truncate_text(
        self,
        text: str,
        max_tokens: int,
        strategy: str = "end"
    ) -> str:
        """截断文本到指定token数量"""
        tokens = self.encoding.encode(text)

        if len(tokens) <= max_tokens:
            return text

        if strategy == "start":
            # 保留开头
            truncated_tokens = tokens[:max_tokens]
        elif strategy == "end":
            # 保留结尾
            truncated_tokens = tokens[-max_tokens:]
        elif strategy == "middle":
            # 保留开头和结尾
            start_tokens = max_tokens // 2
            end_tokens = max_tokens - start_tokens
            truncated_tokens = tokens[:start_tokens] + tokens[-end_tokens:]
        else:
            truncated_tokens = tokens[:max_tokens]

        return self.encoding.decode(truncated_tokens)

    def optimize_prompt(
        self,
        system_prompt: str,
        user_prompt: str,
        context: List[str],
        max_context_ratio: float = 0.6
    ) -> str:
        """优化提示以适应token限制"""

        # 计算各部分token数
        system_tokens = self.count_tokens(system_prompt)
        user_tokens = self.count_tokens(user_prompt)

        # 为响应预留token
        reserved_tokens = self.max_tokens // 4
        available_tokens = self.max_tokens - reserved_tokens - system_tokens - user_tokens

        # 计算上下文可用token
        context_tokens = int(available_tokens * max_context_ratio)

        # 优化上下文
        optimized_context = self._optimize_context(context, context_tokens)

        # 构建最终提示
        final_prompt = f"{system_prompt}\n\nContext:\n{optimized_context}\n\nUser: {user_prompt}"

        return final_prompt