DeerFlow LLMs模块实现逻辑分析

概述

DeerFlow的LLMs模块是整个系统的语言模型管理核心，采用工厂模式和缓存机制，提供统一的多提供商LLM接口。该模块支持OpenAI、Azure OpenAI、Google AI Studio、DashScope、DeepSeek等多种LLM提供商。
模型接入方式，除了几种特殊的外，默认使用OpenAI API方式接入。
项目地址：https://github.com/bytedance/deer-flow
LangGraph中文在线文档：https://github.langchain.ac.cn/langgraph/agents/agents/

1. 核心架构设计

1.1 类型定义与映射

# LLM类型定义
LLMType = Literal["basic", "reasoning", "vision", "code"]

# 配置键映射
def _get_llm_type_config_keys() -> dict[str, str]:
    return {
        "reasoning": "REASONING_MODEL",  # 推理模型
        "basic": "BASIC_MODEL",         # 基础模型
        "vision": "VISION_MODEL",       # 视觉模型
        "code": "CODE_MODEL",           # 代码模型
    }

1.2 缓存机制

# 全局LLM实例缓存
_llm_cache: dict[LLMType, BaseChatModel] = {}

def get_llm_by_type(llm_type: LLMType) -> BaseChatModel:
    """获取LLM实例，优先使用缓存"""
    if llm_type in _llm_cache:
        return _llm_cache[llm_type]
    
    conf = load_yaml_config(_get_config_file_path())
    llm = _create_llm_use_conf(llm_type, conf)
    _llm_cache[llm_type] = llm
    return llm

2. 配置管理系统

2.1 多层配置优先级

YAML配置文件(次优先级) cong.yaml：
配置BASIC、REASONING、VISION、CODE 四类大模型

ollma提供服务的BASIC大模型配置样例：

# BASIC_MODEL:
#   base_url: "http://localhost:11434/v1"  # Ollama OpenAI compatible endpoint
#   model: "qwen3:14b"  # or "llama3.2", etc.
#   api_key: "ollama"  # Ollama doesn't need real API key
#   max_retries: 3
#   verify_ssl: false  # Local deployment usually doesn't need SSL verification

加载配置详细信息：

def _get_env_llm_conf(llm_type: str) -> Dict[str, Any]:
    """
    从环境变量获取LLM配置
    格式: {LLM_TYPE}_MODEL__{KEY}
    例如: BASIC_MODEL__api_key, BASIC_MODEL__base_url
    """
    prefix = f"{llm_type.upper()}_MODEL__"
    conf = {}
    for key, value in os.environ.items():
        if key.startswith(prefix):
            conf_key = key[len(prefix):].lower()
            conf[conf_key] = value
    return conf

2.2 配置合并策略

def _create_llm_use_conf(llm_type: LLMType, conf: Dict[str, Any]) -> BaseChatModel:
    # 获取YAML配置
    llm_conf = conf.get(config_key, {})
    
    # 获取环境变量配置
    env_conf = _get_env_llm_conf(llm_type)
    
    # 合并配置，环境变量优先
    merged_conf = {**llm_conf, **env_conf}
    
    # 移除初始化时不需要的参数
    if "token_limit" in merged_conf:
        merged_conf.pop("token_limit")

3. 多提供商支持

3.1 智能提供商识别

系统通过配置参数自动识别并创建相应的LLM实例：

def _create_llm_use_conf(llm_type: LLMType, conf: Dict[str, Any]) -> BaseChatModel:
    # Google AI Studio检测
    platform = merged_conf.get("platform", "").lower()
    is_google_aistudio = platform == "google_aistudio"
    
    if is_google_aistudio:
        # 处理Google AI Studio特定配置
        if "api_key" in gemini_conf:
            gemini_conf["google_api_key"] = gemini_conf.pop("api_key")
        gemini_conf.pop("base_url", None)
        gemini_conf.pop("platform", None)
        return ChatGoogleGenerativeAI(**gemini_conf)
    
    # Azure OpenAI检测
    if "azure_endpoint" in merged_conf or os.getenv("AZURE_OPENAI_ENDPOINT"):
        return AzureChatOpenAI(**merged_conf)
    
    # DashScope检测
    if "base_url" in merged_conf and "dashscope." in merged_conf["base_url"]:
        if llm_type == "reasoning":
            merged_conf["extra_body"] = {"enable_thinking": True}
        else:
            merged_conf["extra_body"] = {"enable_thinking": False}
        return ChatDashscope(**merged_conf)
    
    # DeepSeek推理模型
    if llm_type == "reasoning":
        merged_conf["api_base"] = merged_conf.pop("base_url", None)
        return ChatDeepSeek(**merged_conf)
    
    # 默认OpenAI
    return ChatOpenAI(**merged_conf)

3.2 SSL验证处理

# 处理SSL验证设置
verify_ssl = merged_conf.pop("verify_ssl", True)

if not verify_ssl:
    http_client = httpx.Client(verify=False)
    http_async_client = httpx.AsyncClient(verify=False)
    merged_conf["http_client"] = http_client
    merged_conf["http_async_client"] = http_async_client

4. DashScope扩展实现

4.1 ChatDashscope类设计

DashScope提供商通过扩展ChatOpenAI实现，支持推理模型的特殊功能：

class ChatDashscope(ChatOpenAI):
    """扩展的ChatOpenAI模型，支持推理功能"""
    
    def _create_chat_result(self, response, generation_info=None) -> ChatResult:
        chat_result = super()._create_chat_result(response, generation_info)
        
        # 提取推理内容
        try:
            if (hasattr(response, "choices") and response.choices and 
                hasattr(response.choices[0], "message") and
                hasattr(response.choices[0].message, "reasoning_content")):
                
                reasoning_content = response.choices[0].message.reasoning_content
                if reasoning_content and chat_result.generations:
                    chat_result.generations[0].message.additional_kwargs[
                        "reasoning_content"
                    ] = reasoning_content
        except (IndexError, AttributeError):
            pass
        
        return chat_result

4.2 流式响应处理

def _convert_delta_to_message_chunk(delta_dict, default_class):
    """将delta字典转换为消息块"""
    message_id = delta_dict.get("id")
    role = delta_dict.get("role", "")
    content = delta_dict.get("content") or ""
    additional_kwargs = {}
    
    # 处理推理内容
    if reasoning_content := delta_dict.get("reasoning_content"):
        additional_kwargs["reasoning_content"] = reasoning_content
    
    # 处理工具调用
    tool_call_chunks = []
    if raw_tool_calls := delta_dict.get("tool_calls"):
        additional_kwargs["tool_calls"] = raw_tool_calls
        try:
            tool_call_chunks = [
                tool_call_chunk(
                    name=rtc.get("function", {}).get("name"),
                    args=rtc.get("function", {}).get("arguments"),
                    id=rtc.get("id"),
                    index=rtc.get("index", 0),
                )
                for rtc in raw_tool_calls
                if rtc.get("function")
            ]
        except (KeyError, TypeError):
            pass
    
    # 根据角色返回相应的消息块
    if role == "assistant":
        return AIMessageChunk(
            content=content,
            additional_kwargs=additional_kwargs,
            id=message_id,
            tool_call_chunks=tool_call_chunks,
        )

4.3 流式生成优化

def _convert_chunk_to_generation_chunk(chunk, default_chunk_class, base_generation_info):
    """将流式块转换为生成块"""
    # 跳过content.delta类型的块
    if chunk.get("type") == "content.delta":
        return None
    
    token_usage = chunk.get("usage")
    choices = chunk.get("choices", []) or chunk.get("chunk", {}).get("choices", [])
    
    # 创建使用元数据
    usage_metadata = _create_usage_metadata(token_usage) if token_usage else None
    
    # 处理空选择
    if not choices:
        return ChatGenerationChunk(
            message=default_chunk_class(content="", usage_metadata=usage_metadata)
        )
    
    choice = choices[0]
    if choice.get("delta") is None:
        return None
    
    # 转换消息块
    message_chunk = _convert_delta_to_message_chunk(choice["delta"], default_chunk_class)
    
    # 添加生成信息
    generation_info = dict(base_generation_info) if base_generation_info else {}
    if finish_reason := choice.get("finish_reason"):
        generation_info["finish_reason"] = finish_reason
    
    # 附加使用元数据到AI消息块
    if usage_metadata and isinstance(message_chunk, AIMessageChunk):
        message_chunk.usage_metadata = usage_metadata
    
    return ChatGenerationChunk(
        message=message_chunk, 
        generation_info=generation_info or None
    )

5. 工具函数

5.1 配置查询

def get_configured_llm_models() -> dict[str, list[str]]:
    """获取所有已配置的LLM模型"""
    try:
        conf = load_yaml_config(_get_config_file_path())
        llm_type_config_keys = _get_llm_type_config_keys()
        configured_models = {}
        
        for llm_type in get_args(LLMType):
            config_key = llm_type_config_keys.get(llm_type, "")
            yaml_conf = conf.get(config_key, {}) if config_key else {}
            env_conf = _get_env_llm_conf(llm_type)
            merged_conf = {**yaml_conf, **env_conf}
            
            model_name = merged_conf.get("model")
            if model_name:
                configured_models.setdefault(llm_type, []).append(model_name)
        
        return configured_models
    except Exception as e:
        print(f"Warning: Failed to load LLM configuration: {e}")
        return {}

5.2 Token限制查询

def get_llm_token_limit_by_type(llm_type: str) -> int:
    """获取指定LLM类型的最大token限制"""
    llm_type_config_keys = _get_llm_type_config_keys()
    config_key = llm_type_config_keys.get(llm_type)
    
    conf = load_yaml_config(_get_config_file_path())
    llm_max_token = conf.get(config_key, {}).get("token_limit")
    return llm_max_token

6. 错误处理与容错

6.1 重试机制

# 添加重试机制处理速率限制错误
if "max_retries" not in merged_conf:
    merged_conf["max_retries"] = 3

6.2 配置验证

if not merged_conf:
    raise ValueError(f"No configuration found for LLM type: {llm_type}")

if not isinstance(llm_conf, dict):
    raise ValueError(f"Invalid LLM configuration for {llm_type}: {llm_conf}")

7. 架构优势

7.1 统一接口

• 所有LLM提供商通过统一的get_llm_by_type()接口访问
• 屏蔽不同提供商的API差异

7.2 配置灵活性

• 支持环境变量和YAML文件双重配置
• 环境变量优先级更高，便于部署时覆盖

7.3 性能优化

• 实例缓存避免重复初始化
• 懒加载机制，按需创建LLM实例

7.4 扩展性

• 新增提供商只需在_create_llm_use_conf中添加识别逻辑
• 支持自定义HTTP客户端和SSL配置

8. 使用示例

# 获取基础模型
basic_llm = get_llm_by_type("basic")

# 获取推理模型
reasoning_llm = get_llm_by_type("reasoning")

# 获取视觉模型
vision_llm = get_llm_by_type("vision")

# 查询配置的模型
models = get_configured_llm_models()
print(models)  # {'basic': ['gpt-4'], 'reasoning': ['deepseek-reasoner']}

# 获取token限制
token_limit = get_llm_token_limit_by_type("basic")

总结

DeerFlow的LLMs模块通过精心设计的架构实现了：

1. 多提供商统一管理：支持主流LLM提供商
2. 灵活的配置系统：环境变量与配置文件双重支持
3. 高性能缓存机制：避免重复初始化开销
4. 强大的扩展能力：易于添加新的LLM提供商
5. 完善的错误处理：重试机制和配置验证

这种设计使得系统能够灵活适应不同的部署环境和LLM提供商需求。