LangChain 源码剖析 - 关键数据结构与UML图

classDiagram
    %% 核心基类
    class Runnable {
        <<interface>>
        +invoke(input, config) Output
        +ainvoke(input, config) Output
        +batch(inputs, config) List[Output]
        +stream(input, config) Iterator[Output]
        +with_config(**kwargs) RunnableBinding
        +with_retry(**kwargs) RunnableRetry
        +with_fallbacks(fallbacks) RunnableWithFallbacks
    }
    
    %% 配置系统
    class RunnableConfig {
        +callbacks: List[BaseCallbackHandler]
        +tags: List[str]
        +metadata: Dict[str, Any]
        +run_name: str
        +run_id: UUID
        +max_concurrency: int
        +recursion_limit: int
        +configurable: Dict[str, Any]
    }
    
    %% 消息系统
    class BaseMessage {
        <<abstract>>
        +content: Union[str, List]
        +additional_kwargs: Dict[str, Any]
        +response_metadata: Dict[str, Any]
        +name: Optional[str]
        +id: Optional[str]
        +type: str
        +__add__(other) BaseMessageChunk
    }
    
    class HumanMessage {
        +type = "human"
    }
    
    class AIMessage {
        +type = "ai"
        +tool_calls: List[ToolCall]
        +invalid_tool_calls: List[InvalidToolCall]
    }
    
    class SystemMessage {
        +type = "system"
    }
    
    class ToolMessage {
        +type = "tool"
        +tool_call_id: str
    }
    
    %% 工具调用
    class ToolCall {
        +name: str
        +args: Dict[str, Any]
        +id: str
    }
    
    class InvalidToolCall {
        +name: Optional[str]
        +args: Optional[str]
        +id: Optional[str]
        +error: Optional[str]
    }
    
    %% 输出结构
    class ChatResult {
        +generations: List[ChatGeneration]
        +llm_output: Optional[Dict[str, Any]]
        +run: Optional[RunInfo]
    }
    
    class ChatGeneration {
        +message: BaseMessage
        +generation_info: Optional[Dict[str, Any]]
    }
    
    class LLMResult {
        +generations: List[List[Generation]]
        +llm_output: Optional[Dict[str, Any]]
        +run: Optional[RunInfo]
    }
    
    class Generation {
        +text: str
        +generation_info: Optional[Dict[str, Any]]
    }
    
    %% 关系定义
    Runnable --> RunnableConfig : uses
    BaseMessage <|-- HumanMessage
    BaseMessage <|-- AIMessage
    BaseMessage <|-- SystemMessage
    BaseMessage <|-- ToolMessage
    AIMessage --> ToolCall : contains
    AIMessage --> InvalidToolCall : contains
    ChatResult --> ChatGeneration : contains
    ChatGeneration --> BaseMessage : contains
    LLMResult --> Generation : contains

classDiagram
    class BaseMessage {
        <<abstract>>
        #content: Union[str, List[Union[str, Dict]]]
        #additional_kwargs: Dict[str, Any]
        #response_metadata: Dict[str, Any] 
        #name: Optional[str]
        #id: Optional[str]
        +type: str*
        +is_lc_serializable() bool
        +__add__(other) BaseMessageChunk
        +_merge_content(left, right) Any
        +_merge_kwargs(left, right) Dict
        +_merge_metadata(left, right) Dict
    }
    
    class HumanMessage {
        +type = "human"
    }
    
    class AIMessage {
        +type = "ai"
        #tool_calls: List[ToolCall]
        #invalid_tool_calls: List[InvalidToolCall]
    }
    
    class SystemMessage {
        +type = "system"
    }
    
    class ToolMessage {
        +type = "tool"
        #tool_call_id: str
    }
    
    class FunctionMessage {
        +type = "function"
        #name: str
    }
    
    class ChatMessage {
        +type = "chat"
        #role: str
    }
    
    class BaseMessageChunk {
        +get_lc_namespace() List[str]
    }
    
    BaseMessage <|-- HumanMessage
    BaseMessage <|-- AIMessage
    BaseMessage <|-- SystemMessage
    BaseMessage <|-- ToolMessage
    BaseMessage <|-- FunctionMessage
    BaseMessage <|-- ChatMessage
    BaseMessage <|-- BaseMessageChunk
    
    note for BaseMessage : "所有消息类型的基类\n支持内容合并和序列化"
    note for AIMessage : "包含工具调用信息\n支持多模态内容"
    note for ToolMessage : "工具执行结果\n必须包含调用ID"

classDiagram
    class ToolCall {
        +name: str
        +args: Dict[str, Any]
        +id: str
        +type: str = "function"
        +to_json() Dict[str, Any]
        +from_json(data) ToolCall
    }
    
    class InvalidToolCall {
        +name: Optional[str]
        +args: Optional[str]
        +id: Optional[str]
        +error: Optional[str]
        +type: str = "invalid_tool_call"
    }
    
    class ToolCallChunk {
        +name: Optional[str]
        +args: Optional[str]
        +id: Optional[str]
        +index: Optional[int]
        +__add__(other) ToolCallChunk
    }
    
    ToolCall --> ToolCallChunk : streaming
    
    note for ToolCall : "完整的工具调用\n包含所有必要信息"
    note for InvalidToolCall : "解析失败的工具调用\n保留原始数据用于调试"
    note for ToolCallChunk : "流式工具调用块\n支持增量构建"

classDiagram
    class LLMResult {
        +generations: List[List[Generation]]
        +llm_output: Optional[Dict[str, Any]]
        +run: Optional[RunInfo]
        +flatten() List[Generation]
    }
    
    class Generation {
        +text: str
        +generation_info: Optional[Dict[str, Any]]
        +type: str = "Generation"
    }
    
    class ChatResult {
        +generations: List[ChatGeneration]
        +llm_output: Optional[Dict[str, Any]]
        +run: Optional[RunInfo]
    }
    
    class ChatGeneration {
        +message: BaseMessage
        +generation_info: Optional[Dict[str, Any]]
        +text: str
        +type: str = "ChatGeneration"
    }
    
    class ChatGenerationChunk {
        +message: BaseMessageChunk
        +generation_info: Optional[Dict[str, Any]]
        +__add__(other) ChatGenerationChunk
    }
    
    class RunInfo {
        +run_id: UUID
    }
    
    LLMResult --> Generation : contains
    ChatResult --> ChatGeneration : contains
    ChatGeneration --> BaseMessage : contains
    ChatGenerationChunk --> BaseMessageChunk : contains
    LLMResult --> RunInfo : optional
    ChatResult --> RunInfo : optional
    
    note for LLMResult : "传统LLM的输出结果\n支持多个生成候选"
    note for ChatResult : "聊天模型的输出结果\n包含结构化消息"
    note for ChatGenerationChunk : "流式聊天生成块\n支持增量合并"

classDiagram
    class BaseCallbackManager {
        <<abstract>>
        +handlers: List[BaseCallbackHandler]
        +inheritable_handlers: List[BaseCallbackHandler]
        +parent_run_id: Optional[UUID]
        +tags: List[str]
        +inheritable_tags: List[str]
        +metadata: Dict[str, Any]
        +inheritable_metadata: Dict[str, Any]
        +add_handler(handler) void
        +remove_handler(handler) void
        +set_handlers(handlers) void
        +add_tags(tags) void
        +add_metadata(metadata) void
    }
    
    class CallbackManager {
        +on_llm_start(serialized, prompts, **kwargs) CallbackManagerForLLMRun
        +on_chat_model_start(serialized, messages, **kwargs) CallbackManagerForLLMRun
        +on_chain_start(serialized, inputs, **kwargs) CallbackManagerForChainRun
        +on_tool_start(serialized, input_str, **kwargs) CallbackManagerForToolRun
        +on_retriever_start(serialized, query, **kwargs) CallbackManagerForRetrieverRun
    }
    
    class AsyncCallbackManager {
        +on_llm_start(serialized, prompts, **kwargs) AsyncCallbackManagerForLLMRun
        +on_chat_model_start(serialized, messages, **kwargs) AsyncCallbackManagerForLLMRun
        +on_chain_start(serialized, inputs, **kwargs) AsyncCallbackManagerForChainRun
        +on_tool_start(serialized, input_str, **kwargs) AsyncCallbackManagerForToolRun
        +on_retriever_start(serialized, query, **kwargs) AsyncCallbackManagerForRetrieverRun
    }
    
    class CallbackManagerForLLMRun {
        +on_llm_new_token(token, **kwargs) void
        +on_llm_end(response, **kwargs) void
        +on_llm_error(error, **kwargs) void
        +get_child(suffix) CallbackManagerForChainRun
    }
    
    class CallbackManagerForChainRun {
        +on_chain_end(outputs, **kwargs) void
        +on_chain_error(error, **kwargs) void
        +on_agent_action(action, **kwargs) void
        +on_agent_finish(finish, **kwargs) void
        +get_child(suffix) CallbackManagerForChainRun
    }
    
    class CallbackManagerForToolRun {
        +on_tool_end(output, **kwargs) void
        +on_tool_error(error, **kwargs) void
        +get_child(suffix) CallbackManagerForChainRun
    }
    
    BaseCallbackManager <|-- CallbackManager
    BaseCallbackManager <|-- AsyncCallbackManager
    CallbackManager --> CallbackManagerForLLMRun : creates
    CallbackManager --> CallbackManagerForChainRun : creates
    CallbackManager --> CallbackManagerForToolRun : creates

classDiagram
    class BasePromptTemplate {
        <<abstract>>
        +input_variables: List[str]
        +input_types: Dict[str, Any]
        +output_parser: Optional[BaseOutputParser]
        +partial_variables: Dict[str, Any]
        +metadata: Optional[Dict[str, Any]]
        +tags: Optional[List[str]]
        +format(**kwargs) str
        +format_prompt(**kwargs) PromptValue
        +partial(**kwargs) BasePromptTemplate
        +save(file_path) void
        +dict() Dict[str, Any]
    }
    
    class PromptTemplate {
        +template: str
        +template_format: str = "f-string"
        +validate_template: bool = False
        +from_template(template) PromptTemplate$
        +from_file(file_path) PromptTemplate$
        +from_examples(examples) PromptTemplate$
    }
    
    class ChatPromptTemplate {
        +messages: List[MessagePromptTemplate]
        +from_template(template) ChatPromptTemplate$
        +from_messages(messages) ChatPromptTemplate$
        +from_role_strings(role_strings) ChatPromptTemplate$
        +append(message) ChatPromptTemplate
        +extend(messages) ChatPromptTemplate
    }
    
    class FewShotPromptTemplate {
        +examples: List[Dict[str, str]]
        +example_prompt: PromptTemplate
        +example_separator: str = "\n\n"
        +prefix: str = ""
        +suffix: str = ""
        +example_selector: Optional[BaseExampleSelector]
        +max_length: Optional[int]
        +get_examples(**kwargs) List[Dict[str, str]]
    }
    
    class PipelinePromptTemplate {
        +final_prompt: BasePromptTemplate
        +pipeline_prompts: List[Tuple[str, BasePromptTemplate]]
        +format(**kwargs) str
    }
    
    BasePromptTemplate <|-- PromptTemplate
    BasePromptTemplate <|-- ChatPromptTemplate
    BasePromptTemplate <|-- FewShotPromptTemplate
    BasePromptTemplate <|-- PipelinePromptTemplate
    
    note for PromptTemplate : "字符串模板\n支持f-string和jinja2格式"
    note for ChatPromptTemplate : "聊天消息模板\n支持角色和系统消息"
    note for FewShotPromptTemplate : "少样本学习模板\n包含示例和选择器"

classDiagram
    class MessagePromptTemplate {
        <<abstract>>
        +format(**kwargs) BaseMessage
        +format_messages(**kwargs) List[BaseMessage]
        +input_variables: List[str]
        +additional_kwargs: Dict[str, Any]
    }
    
    class HumanMessagePromptTemplate {
        +prompt: PromptTemplate
        +additional_kwargs: Dict[str, Any]
        +from_template(template) HumanMessagePromptTemplate$
    }
    
    class AIMessagePromptTemplate {
        +prompt: PromptTemplate
        +additional_kwargs: Dict[str, Any]
        +from_template(template) AIMessagePromptTemplate$
    }
    
    class SystemMessagePromptTemplate {
        +prompt: PromptTemplate
        +additional_kwargs: Dict[str, Any]
        +from_template(template) SystemMessagePromptTemplate$
    }
    
    class ChatMessagePromptTemplate {
        +prompt: PromptTemplate
        +role: str
        +additional_kwargs: Dict[str, Any]
        +from_template(template, role) ChatMessagePromptTemplate$
    }
    
    class MessagesPlaceholder {
        +variable_name: str
        +optional: bool = False
        +format_messages(**kwargs) List[BaseMessage]
    }
    
    MessagePromptTemplate <|-- HumanMessagePromptTemplate
    MessagePromptTemplate <|-- AIMessagePromptTemplate
    MessagePromptTemplate <|-- SystemMessagePromptTemplate
    MessagePromptTemplate <|-- ChatMessagePromptTemplate
    MessagePromptTemplate <|-- MessagesPlaceholder
    
    ChatPromptTemplate --> MessagePromptTemplate : contains
    
    note for MessagesPlaceholder : "消息占位符\n用于插入动态消息列表"

classDiagram
    class Document {
        +page_content: str
        +metadata: Dict[str, Any]
        +id: Optional[str]
        +__init__(page_content, metadata, **kwargs)
        +__str__() str
        +__repr__() str
        +dict() Dict[str, Any]
        +copy(*, deep=True) Document
    }
    
    class DocumentChunk {
        +page_content: str
        +metadata: Dict[str, Any]
        +start_index: Optional[int]
        +end_index: Optional[int]
        +chunk_id: Optional[str]
        +parent_id: Optional[str]
    }
    
    class Embedding {
        +vector: List[float]
        +dimension: int
        +model: Optional[str]
        +normalize() Embedding
        +similarity(other) float
        +distance(other) float
    }
    
    class VectorStoreDocument {
        +id: str
        +vector: List[float]
        +document: Document
        +score: Optional[float]
        +similarity_search_result: bool = False
    }
    
    Document <|-- DocumentChunk
    Document --> VectorStoreDocument : contains
    Embedding --> VectorStoreDocument : contains
    
    note for Document : "基础文档类\n包含内容和元数据"
    note for DocumentChunk : "文档分块\n记录位置信息"
    note for VectorStoreDocument : "向量存储中的文档\n包含向量和相似度"

# 向量存储的详细接口定义
from typing import List, Optional, Dict, Any, Tuple, Iterator
from abc import ABC, abstractmethod

class VectorStore(ABC):
    """
    向量存储抽象基类
    
    定义了向量存储的标准接口，包括：
    - 文档添加和删除
    - 相似性搜索
    - 最大边际相关性搜索
    - 检索器接口
    """
    
    @abstractmethod
    def add_texts(
        self,
        texts: Iterable[str],
        metadatas: Optional[List[Dict[str, Any]]] = None,
        **kwargs: Any,
    ) -> List[str]:
        """
        添加文本到向量存储
        
        Args:
            texts: 文本内容迭代器
            metadatas: 对应的元数据列表
            **kwargs: 额外参数
            
        Returns:
            文档ID列表
        """
        pass
    
    def add_documents(
        self,
        documents: List[Document],
        **kwargs: Any,
    ) -> List[str]:
        """
        添加文档到向量存储
        
        Args:
            documents: 文档列表
            **kwargs: 额外参数
            
        Returns:
            文档ID列表
        """
        texts = [doc.page_content for doc in documents]
        metadatas = [doc.metadata for doc in documents]
        return self.add_texts(texts, metadatas, **kwargs)
    
    @abstractmethod
    def similarity_search(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> List[Document]:
        """
        相似性搜索
        
        Args:
            query: 查询文本
            k: 返回结果数量
            **kwargs: 额外参数
            
        Returns:
            最相似的文档列表
        """
        pass
    
    def similarity_search_with_score(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> List[Tuple[Document, float]]:
        """
        带相似度分数的搜索
        
        Args:
            query: 查询文本
            k: 返回结果数量
            **kwargs: 额外参数
            
        Returns:
            (文档, 相似度分数) 元组列表
        """
        docs = self.similarity_search(query, k=k, **kwargs)
        return [(doc, 1.0) for doc in docs]  # 默认分数为1.0
    
    def similarity_search_by_vector(
        self,
        embedding: List[float],
        k: int = 4,
        **kwargs: Any,
    ) -> List[Document]:
        """
        基于向量的相似性搜索
        
        Args:
            embedding: 查询向量
            k: 返回结果数量
            **kwargs: 额外参数
            
        Returns:
            最相似的文档列表
        """
        raise NotImplementedError("子类必须实现此方法")
    
    def max_marginal_relevance_search(
        self,
        query: str,
        k: int = 4,
        fetch_k: int = 20,
        lambda_mult: float = 0.5,
        **kwargs: Any,
    ) -> List[Document]:
        """
        最大边际相关性搜索
        
        在相似性和多样性之间平衡的搜索算法
        
        Args:
            query: 查询文本
            k: 返回结果数量
            fetch_k: 候选文档数量
            lambda_mult: 多样性权重 (0=最大多样性, 1=最小多样性)
            **kwargs: 额外参数
            
        Returns:
            搜索结果文档列表
        """
        raise NotImplementedError("子类必须实现此方法")
    
    def delete(
        self,
        ids: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> Optional[bool]:
        """
        删除文档
        
        Args:
            ids: 要删除的文档ID列表
            **kwargs: 额外参数
            
        Returns:
            删除成功返回True，否则返回False
        """
        raise NotImplementedError("删除操作未实现")
    
    def as_retriever(self, **kwargs: Any) -> "VectorStoreRetriever":
        """
        转换为检索器接口
        
        Args:
            **kwargs: 检索器参数
            
        Returns:
            VectorStoreRetriever实例
        """
        from langchain_core.vectorstores import VectorStoreRetriever
        
        return VectorStoreRetriever(
            vectorstore=self,
            **kwargs,
        )

class VectorStoreRetriever(BaseRetriever):
    """
    向量存储检索器
    
    将向量存储包装为Retriever接口
    """
    
    vectorstore: VectorStore
    """底层向量存储"""
    
    search_type: str = "similarity"
    """搜索类型：similarity, mmr, similarity_score_threshold"""
    
    search_kwargs: Dict[str, Any] = Field(default_factory=dict)
    """搜索参数"""
    
    def get_relevant_documents(
        self,
        query: str,
        *,
        callbacks: Optional[Callbacks] = None,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        run_name: Optional[str] = None,
        **kwargs: Any,
    ) -> List[Document]:
        """
        获取相关文档
        
        Args:
            query: 查询字符串
            callbacks: 回调处理器
            tags: 标签
            metadata: 元数据
            run_name: 运行名称
            **kwargs: 额外参数
            
        Returns:
            相关文档列表
        """
        if self.search_type == "similarity":
            docs = self.vectorstore.similarity_search(query, **self.search_kwargs)
        elif self.search_type == "mmr":
            docs = self.vectorstore.max_marginal_relevance_search(
                query, **self.search_kwargs
            )
        elif self.search_type == "similarity_score_threshold":
            docs_and_similarities = (
                self.vectorstore.similarity_search_with_relevance_scores(
                    query, **self.search_kwargs
                )
            )
            docs = [doc for doc, _ in docs_and_similarities]
        else:
            raise ValueError(f"search_type of {self.search_type} not allowed.")
        
        return docs