LangChain 架构深度解析：从核心组件到执行流程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class OptimizedLLM:
    """优化的 LLM 调用封装"""

    def __init__(self, base_llm: BaseLanguageModel):
        self.base_llm = base_llm
        self.cache = {}  # 简单内存缓存
        self.request_queue = asyncio.Queue()
        self.batch_size = 10
        self.batch_timeout = 0.1  # 100ms

    async def batch_generate(self, prompts: List[str]) -> List[str]:
        """批量生成 - 提高吞吐量"""
        if len(prompts) == 1:
            return await self.base_llm.agenerate([prompts])

        # 分批处理
        results = []
        for i in range(0, len(prompts), self.batch_size):
            batch = prompts[i:i + self.batch_size]
            batch_results = await self.base_llm.agenerate(batch)
            results.extend(batch_results)

        return results

    def cached_generate(self, prompt: str) -> str:
        """缓存生成 - 避免重复调用"""
        prompt_hash = hashlib.md5(prompt.encode()).hexdigest()

        if prompt_hash in self.cache:
            return self.cache[prompt_hash]

        result = self.base_llm.generate([prompt])
        self.cache[prompt_hash] = result.generations[0][0].text

        return self.cache[prompt_hash]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

class OptimizedRetriever:
    """优化的检索器"""

    def __init__(self, vectorstore: VectorStore):
        self.vectorstore = vectorstore
        self.query_cache = {}
        self.embedding_cache = {}

    def retrieve_with_cache(self, query: str, k: int = 4) -> List[Document]:
        """带缓存的检索"""
        cache_key = f"{query}:{k}"

        if cache_key in self.query_cache:
            return self.query_cache[cache_key]

        docs = self.vectorstore.similarity_search(query, k=k)
        self.query_cache[cache_key] = docs

        return docs

    def hybrid_search(self, query: str, k: int = 4, alpha: float = 0.7) -> List[Document]:
        """混合检索 - 向量 + BM25"""
        # 向量检索
        vector_docs = self.vectorstore.similarity_search(query, k=k*2)

        # BM25 检索 (简化实现)
        bm25_docs = self._bm25_search(query, k=k*2)

        # 结果融合
        combined_docs = self._combine_results(vector_docs, bm25_docs, alpha)

        return combined_docs[:k]

    def _combine_results(self, vector_docs: List[Document], bm25_docs: List[Document], alpha: float) -> List[Document]:
        """结果融合算法"""
        # 实现 RRF (Reciprocal Rank Fusion) 或其他融合策略
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class RobustChain(Chain):
    """健壮的链实现 - 带重试和降级"""

    def __init__(self, base_chain: Chain, max_retries: int = 3, fallback_chain: Optional[Chain] = None):
        self.base_chain = base_chain
        self.max_retries = max_retries
        self.fallback_chain = fallback_chain

    def _call_with_retry(self, inputs: Dict[str, Any], run_manager: Optional[CallbackManagerForChainRun] = None) -> Dict[str, Any]:
        """带重试的调用"""
        last_exception = None

        for attempt in range(self.max_retries + 1):
            try:
                return self.base_chain(inputs, callbacks=run_manager.get_child() if run_manager else None)

            except Exception as e:
                last_exception = e

                if attempt < self.max_retries:
                    # 指数退避
                    wait_time = (2 ** attempt) * 0.1
                    time.sleep(wait_time)

                    if run_manager:
                        run_manager.on_text(f"Retry attempt {attempt + 1} after error: {str(e)}")
                else:
                    # 最后一次尝试失败，使用降级策略
                    if self.fallback_chain:
                        if run_manager:
                            run_manager.on_text("Using fallback chain due to repeated failures")
                        return self.fallback_chain(inputs, callbacks=run_manager.get_child() if run_manager else None)

        # 所有重试都失败
        raise last_exception

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

class MonitoringCallbackHandler(BaseCallbackHandler):
    """监控回调处理器"""

    def __init__(self, metrics_client=None, alert_thresholds=None):
        self.metrics_client = metrics_client
        self.alert_thresholds = alert_thresholds or {
            "llm_duration": 30.0,  # 30秒
            "error_rate": 0.05,    # 5%
            "token_usage": 10000   # 10k tokens
        }
        self.error_count = 0
        self.total_requests = 0

    def on_llm_end(self, response: LLMResult, *, run_id: UUID, **kwargs) -> None:
        """监控 LLM 性能"""
        self.total_requests += 1

        # 记录 Token 使用
        if response.llm_output and "token_usage" in response.llm_output:
            tokens = response.llm_output["token_usage"]["total_tokens"]

            if self.metrics_client:
                self.metrics_client.gauge("llm.token_usage", tokens)

            # 检查告警阈值
            if tokens > self.alert_thresholds["token_usage"]:
                self._send_alert(f"High token usage: {tokens}")

    def on_llm_error(self, error: Union[Exception, KeyboardInterrupt], *, run_id: UUID, **kwargs) -> None:
        """监控错误率"""
        self.error_count += 1
        error_rate = self.error_count / max(self.total_requests, 1)

        if self.metrics_client:
            self.metrics_client.gauge("llm.error_rate", error_rate)

        if error_rate > self.alert_thresholds["error_rate"]:
            self._send_alert(f"High error rate: {error_rate:.2%}")

    def _send_alert(self, message: str):
        """发送告警"""
        logger.error(f"ALERT: {message}")
        # 集成 Slack、邮件等告警渠道

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

class LangChainConfig:
    """LangChain 配置管理"""

    def __init__(self):
        self.llm_config = {
            "model": os.getenv("LLM_MODEL", "gpt-3.5-turbo"),
            "temperature": float(os.getenv("LLM_TEMPERATURE", "0.1")),
            "max_tokens": int(os.getenv("LLM_MAX_TOKENS", "1000")),
            "timeout": int(os.getenv("LLM_TIMEOUT", "30")),
            "max_retries": int(os.getenv("LLM_MAX_RETRIES", "3"))
        }

        self.agent_config = {
            "max_iterations": int(os.getenv("AGENT_MAX_ITERATIONS", "15")),
            "max_execution_time": int(os.getenv("AGENT_MAX_EXECUTION_TIME", "120")),
            "verbose": os.getenv("AGENT_VERBOSE", "false").lower() == "true"
        }

        self.retrieval_config = {
            "chunk_size": int(os.getenv("CHUNK_SIZE", "1000")),
            "chunk_overlap": int(os.getenv("CHUNK_OVERLAP", "200")),
            "top_k": int(os.getenv("RETRIEVAL_TOP_K", "4")),
            "similarity_threshold": float(os.getenv("SIMILARITY_THRESHOLD", "0.7"))
        }

    def create_llm(self) -> BaseLanguageModel:
        """创建配置化的 LLM"""
        return ChatOpenAI(**self.llm_config)

    def create_agent_executor(self, agent: Runnable, tools: List[BaseTool]) -> AgentExecutor:
        """创建配置化的 Agent 执行器"""
        return AgentExecutor(
            agent=agent,
            tools=tools,
            **self.agent_config
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

// 核心抽象层
from langchain_core.runnables import Runnable
from langchain_core.language_models import BaseLanguageModel
from langchain_core.tools import BaseTool
from langchain_core.memory import BaseMemory

// 实现层 - 可插拔组件
class PluggableAgent:
    """可插拔的 Agent 架构"""

    def __init__(self,
                 llm: BaseLanguageModel,
                 tools: List[BaseTool],
                 memory: BaseMemory,
                 planner: BasePlanner,
                 executor: BaseExecutor):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.planner = planner
        self.executor = executor

    def run(self, query: str) -> str:
        """执行 Agent 任务"""
        # 1. 加载上下文
        context = self.memory.load_context()

        # 2. 规划任务
        plan = self.planner.plan(query, context, self.tools)

        # 3. 执行计划
        result = self.executor.execute(plan, self.tools, self.llm)

        # 4. 保存结果
        self.memory.save_context(query, result)

        return result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class DistributedAgent:
    """分布式 Agent 系统"""

    def __init__(self, node_id: str, cluster_config: Dict):
        self.node_id = node_id
        self.cluster_config = cluster_config
        self.message_queue = MessageQueue(cluster_config["mq_url"])
        self.state_store = StateStore(cluster_config["state_url"])

    async def coordinate_task(self, task: Task) -> TaskResult:
        """协调分布式任务执行"""
        # 1. 任务分解
        subtasks = self.decompose_task(task)

        # 2. 分发子任务
        futures = []
        for subtask in subtasks:
            node = self.select_node(subtask)
            future = self.delegate_task(node, subtask)
            futures.append(future)

        # 3. 收集结果
        results = await asyncio.gather(*futures)

        # 4. 结果聚合
        final_result = self.aggregate_results(results)

        return final_result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

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安全管理器

    参考：LangChain解析器与下游任务集成的源码实现剖析
    https://blog.csdn.net/qq_28540861/article/details/149057817
    """

    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

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)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

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