LanguageModelInput = PromptValue | str | Sequence[MessageLikeRepresentation]

LanguageModelOutput = BaseMessage | str

flowchart TB
    subgraph UserLayer["用户层"]
        USER[应用代码]
        CHAIN[Chains/Agents]
    end

    subgraph InterfaceLayer["接口层"]
        BASE_LM[BaseLanguageModel<br/>统一抽象接口]
        RUNNABLE[RunnableSerializable<br/>可组合执行单元]
    end

    subgraph ModelLayer["模型层"]
        BASE_CHAT[BaseChatModel<br/>聊天模型]
        BASE_LLM[BaseLLM<br/>文本模型]
        SIMPLE_CHAT[SimpleChatModel]
        LLM[LLM]
    end

    subgraph ImplementationLayer["实现层"]
        OPENAI[ChatOpenAI]
        ANTHROPIC[ChatAnthropic]
        GEMINI[ChatGemini]
        OLLAMA[ChatOllama]
    end

    subgraph InfrastructureLayer["基础设施层"]
        CACHE[Caches<br/>缓存系统]
        CALLBACK[Callbacks<br/>回调系统]
        MESSAGES[Messages<br/>消息体系]
        OUTPUTS[Outputs<br/>输出封装]
        PROMPTS[Prompts<br/>提示模板]
        TOOLS_MOD[Tools<br/>工具系统]
        PARSERS[Output Parsers<br/>输出解析器]
    end

    subgraph ExternalLayer["外部服务层"]
        OPENAI_API[OpenAI API]
        ANTHROPIC_API[Anthropic API]
        GOOGLE_API[Google API]
        LOCAL_API[本地模型服务]
    end

    USER --> CHAIN
    CHAIN --> BASE_LM
    USER --> BASE_LM

    BASE_LM -.继承.-> RUNNABLE
    BASE_LM --> BASE_CHAT
    BASE_LM --> BASE_LLM

    BASE_CHAT --> SIMPLE_CHAT
    BASE_LLM --> LLM

    BASE_CHAT -.实现.-> OPENAI
    BASE_CHAT -.实现.-> ANTHROPIC
    BASE_CHAT -.实现.-> GEMINI
    BASE_CHAT -.实现.-> OLLAMA

    BASE_CHAT -.依赖.-> CACHE
    BASE_CHAT -.依赖.-> CALLBACK
    BASE_CHAT -.依赖.-> MESSAGES
    BASE_CHAT -.依赖.-> OUTPUTS
    BASE_CHAT -.依赖.-> PROMPTS
    BASE_CHAT -.依赖.-> TOOLS_MOD
    BASE_CHAT -.依赖.-> PARSERS

    OPENAI --> OPENAI_API
    ANTHROPIC --> ANTHROPIC_API
    GEMINI --> GOOGLE_API
    OLLAMA --> LOCAL_API

    style USER fill:#e1f5ff
    style BASE_LM fill:#fff4e1
    style BASE_CHAT fill:#ffe1e1
    style OPENAI fill:#e8f5e9
    style CACHE fill:#f0e1ff
    style CALLBACK fill:#f0e1ff

flowchart LR
    subgraph Core["核心模型模块"]
        BCM[BaseChatModel]
    end

    subgraph Input["输入处理"]
        PROMPT_VAL[PromptValue]
        CONV[convert_to_messages]
        NORM[_normalize_messages]
    end

    subgraph Cache["缓存模块"]
        CACHE_RESOLVE[_resolve_cache]
        CACHE_LOOKUP[lookup]
        CACHE_UPDATE[update]
    end

    subgraph Callback["回调模块"]
        CB_MGR[CallbackManager]
        CB_START[on_chat_model_start]
        CB_TOKEN[on_llm_new_token]
        CB_END[on_llm_end]
        CB_ERROR[on_llm_error]
    end

    subgraph Generation["生成模块"]
        GEN[_generate]
        GEN_CACHE[_generate_with_cache]
        STREAM[_stream]
    end

    subgraph Output["输出处理"]
        CHAT_RES[ChatResult]
        CHAT_GEN[ChatGeneration]
        AI_MSG[AIMessage]
        LLM_RES[LLMResult]
    end

    subgraph External["外部API"]
        API[Provider API]
    end

    BCM --> PROMPT_VAL
    PROMPT_VAL --> CONV
    CONV --> NORM

    BCM --> CACHE_RESOLVE
    CACHE_RESOLVE --> CACHE_LOOKUP
    CACHE_RESOLVE --> CACHE_UPDATE

    BCM --> CB_MGR
    CB_MGR --> CB_START
    CB_MGR --> CB_TOKEN
    CB_MGR --> CB_END
    CB_MGR --> CB_ERROR

    NORM --> GEN_CACHE
    GEN_CACHE --> CACHE_LOOKUP
    GEN_CACHE --> GEN
    GEN_CACHE --> STREAM
    GEN --> API
    STREAM --> API

    GEN --> CHAT_RES
    CHAT_RES --> CHAT_GEN
    CHAT_GEN --> AI_MSG
    CHAT_RES --> CACHE_UPDATE

    AI_MSG --> LLM_RES

classDiagram
    class BaseLanguageModel~Input, Output~ {
        <<abstract>>
        +cache: BaseCache|bool|None
        +callbacks: Callbacks
        +tags: list[str]|None
        +metadata: dict|None
        +generate_prompt(prompts: list[PromptValue]) LLMResult*
        +invoke(input: LanguageModelInput) LanguageModelOutput
        +batch(inputs: list[Input]) list[Output]
        +stream(input: Input) Iterator[Output]
    }

    class BaseChatModel {
        <<abstract>>
        +_generate(messages: list[BaseMessage]) ChatResult*
        +_stream(messages: list[BaseMessage]) Iterator[ChatGenerationChunk]
        +bind_tools(tools: list[Tool]) BaseChatModel
        +with_structured_output(schema: Type[T]) Runnable[Input, T]
        +_generate_with_cache(...) ChatResult
    }

    class SimpleChatModel {
        <<abstract>>
        +_call(messages: list[BaseMessage], stop: list[str]) str*
    }

    class BaseLLM {
        <<abstract>>
        +_generate(prompts: list[str]) LLMResult*
        +_stream(prompt: str) Iterator[GenerationChunk]
    }

    class LLM {
        <<abstract>>
        +_call(prompt: str, stop: list[str]) str*
    }

    class LLMResult {
        +generations: list[list[Generation]]
        +llm_output: dict|None
        +run: list[RunInfo]|None
    }

    class ChatResult {
        +generations: list[ChatGeneration]
        +llm_output: dict|None
    }

    class ChatGeneration {
        +message: AIMessage
        +generation_info: dict|None
    }

    class AIMessage {
        +content: str|list
        +additional_kwargs: dict
        +response_metadata: dict
        +tool_calls: list[ToolCall]
        +usage_metadata: UsageMetadata
    }

    BaseLanguageModel <|-- BaseChatModel
    BaseLanguageModel <|-- BaseLLM
    BaseChatModel <|-- SimpleChatModel
    BaseLLM <|-- LLM

    BaseChatModel --> ChatResult : returns
    ChatResult --> ChatGeneration : contains
    ChatGeneration --> AIMessage : contains
    BaseLLM --> LLMResult : returns

class UsageMetadata(BaseModel):
    input_tokens: int  # 输入token数
    output_tokens: int  # 输出token数
    total_tokens: int  # 总token数

sequenceDiagram
    autonumber
    participant App as 应用代码
    participant Runnable as Runnable接口层
    participant BCM as BaseChatModel
    participant Input as 输入处理
    participant Cache as 缓存层
    participant CB as 回调层
    participant GenCache as _generate_with_cache
    participant RateLimit as 速率限制
    participant Stream as _stream/_generate
    participant Impl as 实现层(ChatOpenAI)
    participant API as 外部API
    participant Output as 输出处理

    App->>Runnable: invoke(input, config)
    Note over Runnable: Runnable统一入口<br/>支持配置、重试、Fallback

    Runnable->>BCM: invoke(input, config)
    Note over BCM: BaseChatModel<br/>核心调度逻辑

    BCM->>Input: _convert_input(input)
    Input->>Input: 规范化为PromptValue
    Input->>Input: to_messages()
    Input->>Input: _normalize_messages()
    Input-->>BCM: 标准化消息列表

    BCM->>BCM: generate_prompt([ChatPromptValue])
    BCM->>BCM: generate(messages_list)

    BCM->>CB: CallbackManager.configure()
    CB->>CB: on_chat_model_start(model, messages)
    CB-->>BCM: run_manager

    BCM->>GenCache: _generate_with_cache(messages)

    GenCache->>Cache: _resolve_cache()
    Cache-->>GenCache: llm_cache

    alt 缓存已启用
        GenCache->>Cache: _get_llm_string(params)
        Cache-->>GenCache: llm_string
        GenCache->>Cache: cache.lookup(prompt, llm_string)

        alt 缓存命中
            Cache-->>GenCache: cached_generations
            GenCache-->>BCM: ChatResult(cached)
            BCM->>CB: on_llm_end(result)
            BCM-->>App: AIMessage
            Note over BCM,App: 缓存命中,直接返回<br/>耗时<10ms
        end
    end

    Note over GenCache: 缓存未命中,继续执行

    GenCache->>RateLimit: rate_limiter.acquire()
    RateLimit-->>GenCache: 继续

    alt 应使用流式输出
        GenCache->>Stream: _stream(messages)
        Stream->>Impl: 子类实现_stream
        Impl->>API: HTTP Stream请求

        loop 每个token
            API-->>Impl: chunk_data
            Impl->>Impl: 解析chunk
            Impl-->>Stream: ChatGenerationChunk
            Stream->>CB: on_llm_new_token(token, chunk)
            Stream->>Stream: 聚合chunks
        end

        API-->>Impl: [DONE]
        Impl-->>Stream: 流结束
        Stream->>Output: generate_from_stream(chunks)
        Output-->>GenCache: ChatResult

    else 非流式输出
        GenCache->>Stream: _generate(messages)
        Stream->>Impl: 子类实现_generate
        Impl->>Impl: 构造API请求
        Impl->>API: HTTP请求
        Note over API: LLM推理<br/>耗时1-10秒
        API-->>Impl: 完整响应
        Impl->>Impl: 解析响应
        Impl->>Output: 构造ChatResult
        Output-->>Stream: ChatResult
        Stream-->>GenCache: ChatResult
    end

    GenCache->>Output: 添加response_metadata
    GenCache->>Output: 设置message.id

    alt 缓存已启用
        GenCache->>Cache: cache.update(prompt, llm_string, generations)
        Cache-->>GenCache: 更新完成
    end

    GenCache-->>BCM: ChatResult
    BCM->>CB: on_llm_end(result)
    BCM->>Output: 提取generations[0][0].message
    Output-->>BCM: AIMessage
    BCM-->>Runnable: AIMessage
    Runnable-->>App: AIMessage

# Runnable基类提供的能力
class RunnableSerializable:
    def invoke(self, input, config=None, **kwargs):
        # 1) 处理配置(重试、fallback、超时等)
        config = ensure_config(config)

        # 2) 应用配置化字段
        bound = self._bind_config(config)

        # 3) 委托给具体实现
        return bound._invoke(input, config, **kwargs)

def _convert_input(self, model_input: LanguageModelInput) -> PromptValue:
    # 1) PromptValue直接返回
    if isinstance(model_input, PromptValue):
        return model_input

    # 2) 字符串包装为StringPromptValue
    if isinstance(model_input, str):
        return StringPromptValue(text=model_input)

    # 3) 消息列表包装为ChatPromptValue
    if isinstance(model_input, Sequence):
        messages = convert_to_messages(model_input)  # 支持dict/tuple转换
        return ChatPromptValue(messages=messages)

    raise ValueError(f"Invalid input type {type(model_input)}")

def to_messages(self) -> list[BaseMessage]:
    # ChatPromptValue: 返回消息列表
    # StringPromptValue: 包装为[HumanMessage(content=text)]
    ...

def _normalize_messages(messages: list[BaseMessage]) -> list[BaseMessage]:
    """规范化消息格式"""
    # 1) 处理多模态内容(图片、音频、文件)
    # 2) 展开MessageLike对象
    # 3) 验证消息顺序和角色
    ...

def _format_for_tracing(messages: list[BaseMessage]) -> list[BaseMessage]:
    """格式化用于追踪的消息"""
    # 1) 转换图片块为OpenAI格式(向后兼容)
    # 2) 为content block添加type字段
    # 3) 浅拷贝消息避免修改原始数据
    ...

"你好"
  → StringPromptValue(text="你好")
  → [HumanMessage(content="你好")]
  → 规范化消息
  → 追踪格式化

[{"role": "user", "content": "你好"}]
  → convert_to_messages
  → [HumanMessage(content="你好")]
  → 规范化消息

[HumanMessage("你好"), AIMessage("你好!"), HumanMessage("天气如何?")]
  → 直接使用
  → 规范化消息

def generate(self, messages: list[list[BaseMessage]], stop=None, callbacks=None, **kwargs):
    # 1) 获取LangSmith追踪参数
    params = self._get_invocation_params(stop=stop, **kwargs)
    ls_params = self._get_ls_params(stop=stop, **kwargs)

    # 2) 合并元数据
    inheritable_metadata = {
        **(metadata or {}),
        **ls_params,  # ls_provider, ls_model_name, ls_temperature等
    }

    # 3) 创建回调管理器
    callback_manager = CallbackManager.configure(
        callbacks,           # 本次调用的回调
        self.callbacks,      # 模型实例的回调
        self.verbose,        # 是否打印详细信息
        tags,                # 标签
        self.tags,           # 模型实例标签
        inheritable_metadata, # 元数据
        self.metadata,       # 模型实例元数据
    )

    # 4) 触发开始回调
    run_managers = callback_manager.on_chat_model_start(
        self._serialized,           # 模型序列化信息
        messages_to_trace,          # 格式化后的消息
        invocation_params=params,   # 调用参数
        options=options,            # 选项
        name=run_name,              # 运行名称
        run_id=run_id,              # 运行ID
        batch_size=len(messages),   # 批次大小
    )

    return run_managers

t0: on_chat_model_start(model, messages, params)
    ↓
t1-t9: [流式模式] on_llm_new_token(token, chunk) × N次
    ↓
t10: on_llm_end(result)  或  on_llm_error(error)

def _generate_with_cache(self, messages, stop=None, run_manager=None, **kwargs):
    # 1) 解析缓存配置
    llm_cache = self.cache if isinstance(self.cache, BaseCache) else get_llm_cache()
    check_cache = self.cache or self.cache is None

    if check_cache and llm_cache:
        # 2) 生成缓存键
        llm_string = self._get_llm_string(stop=stop, **kwargs)
        # llm_string示例: '{"model":"gpt-4o","temperature":0.7}---[("stop",["\\n"])]'

        prompt = dumps(messages)  # 序列化消息列表

        # 3) 查询缓存
        cache_val = llm_cache.lookup(prompt, llm_string)

        if isinstance(cache_val, list):
            # 缓存命中!转换并返回
            converted = self._convert_cached_generations(cache_val)
            return ChatResult(generations=converted)

    # 缓存未命中,继续执行生成逻辑
    ...

cache_key = hash(prompt + llm_string)

prompt = dumps([
    HumanMessage(content="什么是量子计算?")
])
# → "{"type":"human","data":{"content":"什么是量子计算?"}}"

llm_string = JSON({
    "model": "gpt-4o",
    "temperature": 0.7,
    "max_tokens": null,
    ...  # 所有_identifying_params
}) + "---" + str([("stop", ["\\n"])])

# 在缓存查询后,API调用前执行
if self.rate_limiter:
    self.rate_limiter.acquire(blocking=True)

# 使用示例
from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=10,  # 每秒最多10次请求
    check_every_n_seconds=0.1,  # 每0.1秒检查一次
)

model = ChatOpenAI(
    model="gpt-4o",
    rate_limiter=rate_limiter,
)

def _should_stream(self, *, async_api: bool, run_manager=None, **kwargs) -> bool:
    # 1) 检查是否实现了_stream
    sync_not_implemented = type(self)._stream == BaseChatModel._stream
    async_not_implemented = type(self)._astream == BaseChatModel._astream

    if (not async_api) and sync_not_implemented:
        return False
    if async_api and async_not_implemented and sync_not_implemented:
        return False

    # 2) 检查实例级禁用
    if self.disable_streaming is True:
        return False
    if self.disable_streaming == "tool_calling" and kwargs.get("tools"):
        return False  # 工具调用模式下禁用流式

    # 3) 检查运行时参数
    if "stream" in kwargs:
        return kwargs["stream"]

    # 4) 检查模型字段配置
    if "streaming" in self.model_fields_set:
        streaming_value = getattr(self, "streaming", None)
        if isinstance(streaming_value, bool):
            return streaming_value

    # 5) 检查回调处理器
    handlers = run_manager.handlers if run_manager else []
    return any(isinstance(h, _StreamingCallbackHandler) for h in handlers)

是否实现_stream?
  ├─ 否 → 使用非流式
  └─ 是 → disable_streaming?
          ├─ True → 使用非流式
          ├─ "tool_calling" + tools存在 → 使用非流式
          └─ False → kwargs["stream"]?
                     ├─ True → 使用流式
                     ├─ False → 使用非流式
                     └─ 未设置 → streaming字段?
                                 ├─ True → 使用流式
                                 ├─ False → 使用非流式
                                 └─ 未设置 → 有StreamingCallback?
                                             ├─ 是 → 使用流式
                                             └─ 否 → 使用非流式

# BaseChatModel中的流式逻辑
def _generate_with_cache(self, messages, stop=None, run_manager=None, **kwargs):
    if self._should_stream(async_api=False, run_manager=run_manager, **kwargs):
        chunks: list[ChatGenerationChunk] = []
        run_id = f"{LC_ID_PREFIX}-{run_manager.run_id}" if run_manager else None

        # 调用子类实现的_stream
        for chunk in self._stream(messages, stop=stop, **kwargs):
            # 1) 设置响应元数据
            chunk.message.response_metadata = _gen_info_and_msg_metadata(chunk)

            # 2) 处理output_version
            if self.output_version == "v1":
                chunk.message = _update_message_content_to_blocks(chunk.message, "v1")

            # 3) 设置消息ID
            if run_manager and chunk.message.id is None:
                chunk.message.id = run_id

            # 4) 触发token回调
            if run_manager:
                run_manager.on_llm_new_token(
                    cast("str", chunk.message.content),
                    chunk=chunk
                )

            # 5) 收集chunk
            chunks.append(chunk)

        # 6) 从chunks生成最终结果
        result = generate_from_stream(iter(chunks))
        return result

def _stream(self, messages, stop=None, run_manager=None, **kwargs):
    # 1) 构造OpenAI API请求
    request_params = self._prepare_request(messages, stop, stream=True, **kwargs)

    # 2) 发起流式请求
    response = self.client.chat.completions.create(
        **request_params,
        stream=True,
    )

    # 3) 逐chunk处理
    for chunk in response:
        if not chunk.choices:
            continue

        choice = chunk.choices[0]

        # 4) 提取增量内容
        delta = choice.delta
        content = delta.content or ""
        tool_call_chunks = self._extract_tool_call_chunks(delta)

        # 5) 构造消息chunk
        message_chunk = AIMessageChunk(
            content=content,
            tool_call_chunks=tool_call_chunks,
        )

        # 6) 包装为ChatGenerationChunk
        generation_chunk = ChatGenerationChunk(
            message=message_chunk,
            generation_info={
                "finish_reason": choice.finish_reason,
                "model": chunk.model,
            },
        )

        yield generation_chunk

def generate_from_stream(stream: Iterator[ChatGenerationChunk]) -> ChatResult:
    # 1) 读取第一个chunk
    generation = next(stream, None)

    if generation:
        # 2) 累加所有chunk
        for chunk in stream:
            generation += chunk  # AIMessageChunk.__add__实现累加

        # 3) 转换为完整消息
        message = message_chunk_to_message(generation.message)

        return ChatResult(
            generations=[ChatGeneration(
                message=message,
                generation_info=generation.generation_info,
            )]
        )

# 非流式路径
if inspect.signature(self._generate).parameters.get("run_manager"):
    result = self._generate(
        messages,
        stop=stop,
        run_manager=run_manager,
        **kwargs
    )
else:
    # 旧版本不支持run_manager参数
    result = self._generate(messages, stop=stop, **kwargs)

def _generate(self, messages, stop=None, run_manager=None, **kwargs):
    # 1) 构造API请求
    request_params = self._prepare_request(messages, stop, stream=False, **kwargs)

    # 2) 同步调用API
    response = self.client.chat.completions.create(**request_params)

    # 3) 解析响应
    choice = response.choices[0]
    message = AIMessage(
        content=choice.message.content or "",
        tool_calls=self._extract_tool_calls(choice.message),
        response_metadata={
            "model": response.model,
            "finish_reason": choice.finish_reason,
            "system_fingerprint": response.system_fingerprint,
        },
        usage_metadata=UsageMetadata(
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            total_tokens=response.usage.total_tokens,
        ),
    )

    # 4) 包装为ChatResult
    generation = ChatGeneration(
        message=message,
        generation_info={
            "finish_reason": choice.finish_reason,
        },
    )

    return ChatResult(
        generations=[generation],
        llm_output={"model": response.model},
    )

# 1) 处理output_version
if self.output_version == "v1":
    for generation in result.generations:
        generation.message = _update_message_content_to_blocks(
            generation.message, "v1"
        )

# 2) 添加消息ID
for idx, generation in enumerate(result.generations):
    if run_manager and generation.message.id is None:
        generation.message.id = f"{LC_ID_PREFIX}-{run_manager.run_id}-{idx}"

    # 3) 合并响应元数据
    generation.message.response_metadata = _gen_info_and_msg_metadata(generation)

# 4) 特殊处理单个generation的情况
if len(result.generations) == 1 and result.llm_output is not None:
    result.generations[0].message.response_metadata = {
        **result.llm_output,
        **result.generations[0].message.response_metadata,
    }

if check_cache and llm_cache:
    llm_cache.update(prompt, llm_string, result.generations)

# 触发结束回调
run_manager.on_llm_end(LLMResult(generations=generations, llm_output=llm_output))

# 提取AIMessage
message = cast("AIMessage", result.generations[0][0].message)

# 返回给用户
return message

sequenceDiagram
    autonumber
    participant App
    participant BCM as BaseChatModel
    participant Gen as generate()
    participant Cache
    participant Batch as 批处理逻辑
    participant Worker as 工作线程池

    App->>BCM: batch([input1, input2, input3], config)
    BCM->>Gen: generate([msg1, msg2, msg3])

    Gen->>Cache: 批量查询缓存
    Cache-->>Gen: [cached1, miss, cached3]

    Gen->>Batch: 只处理miss的输入

    par 并发执行
        Batch->>Worker: _generate_with_cache(msg2)
        Worker->>Worker: API调用
        Worker-->>Batch: result2
    end

    Batch->>Cache: 更新缓存
    Batch->>Gen: 合并结果[cached1, result2, cached3]
    Gen-->>BCM: LLMResult
    BCM-->>App: [AIMessage1, AIMessage2, AIMessage3]

def generate(self, messages: list[list[BaseMessage]], stop=None, callbacks=None, **kwargs):
    # 1) 缓存批量查询
    existing_prompts, llm_string, missing_prompt_idxs, missing_prompts = \
        get_prompts(params, [dumps(m) for m in messages], self.cache)

    # 2) 只生成缓存未命中的部分
    if len(missing_prompts) > 0:
        run_managers = [
            callback_managers[idx].on_llm_start(...)
            for idx in missing_prompt_idxs
        ]

        # 批量调用_generate (子类可实现真正的批处理)
        new_results = self._generate_helper(
            missing_prompts,
            stop,
            run_managers,
            **kwargs,
        )

        # 3) 更新缓存
        update_cache(self.cache, existing_prompts, llm_string,
                     missing_prompt_idxs, new_results, prompts)

    # 4) 合并结果
    generations = [existing_prompts[i] for i in range(len(prompts))]
    return LLMResult(generations=generations, llm_output=llm_output)

sequenceDiagram
    autonumber
    participant App
    participant BCM as BaseChatModel
    participant AGen as agenerate()
    participant Executor as AsyncExecutor
    participant API as 异步API

    App->>BCM: await ainvoke(input)
    BCM->>AGen: await agenerate([messages])

    alt 子类实现了_agenerate
        AGen->>API: await self._agenerate(messages)
        API-->>AGen: result
    else 只实现了_generate
        AGen->>Executor: run_in_executor(self._generate, messages)
        Note over Executor: 在线程池中执行同步代码
        Executor-->>AGen: result
    end

    AGen-->>BCM: LLMResult
    BCM-->>App: AIMessage

# invoke方法参数
input: LanguageModelInput  # str | list[BaseMessage] | PromptValue
config: RunnableConfig | None  # 可选配置
stop: list[str] | None  # 停止词
**kwargs: Any  # 传递给模型的额外参数(如temperature)

# invoke返回AIMessage
class AIMessage(BaseMessage):
    content: str | list[ContentBlock]  # 生成的内容
    response_metadata: dict  # 响应元数据
    tool_calls: list[ToolCall]  # 工具调用请求
    usage_metadata: UsageMetadata  # Token使用情况

class BaseChatModel(BaseLanguageModel[LanguageModelInput, AIMessage]):
    def invoke(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AIMessage:
        config = ensure_config(config)

        # 1) 转换输入为消息列表
        messages = self._convert_input(input).to_messages()

        # 2) 调用generate_prompt并提取第一个结果
        result = self.generate_prompt(
            [ChatPromptValue(messages=messages)],
            stop=stop,
            callbacks=config.get("callbacks"),
            tags=config.get("tags"),
            metadata=config.get("metadata"),
            run_name=config.get("run_name"),
            **kwargs,
        )

        # 3) 返回第一个生成的消息
        return cast(AIMessage, result.generations[0][0].message)

    def generate_prompt(
        self,
        prompts: list[PromptValue],
        stop: list[str] | None = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> LLMResult:
        # 4) 转换PromptValue为消息列表
        messages_list = [p.to_messages() for p in prompts]

        # 5) 调用generate
        return self.generate(messages_list, stop=stop, callbacks=callbacks, **kwargs)

    def generate(
        self,
        messages: list[list[BaseMessage]],
        stop: list[str] | None = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> LLMResult:
        # 6) 创建回调管理器
        callback_manager = CallbackManager.configure(
            callbacks,
            self.callbacks,
            self.tags,
            self.metadata,
            self.verbose,
        )
        run_manager = callback_manager.on_chat_model_start(
            dumpd(self),
            messages,
        )

        # 7) 批量生成(带缓存)
        results = []
        for m in messages:
            result = self._generate_with_cache(
                m,
                stop=stop,
                run_manager=run_manager,
                **kwargs,
            )
            results.append(result)

        # 8) 合并结果
        llm_output = self._combine_llm_outputs([r.llm_output for r in results])
        generations = [r.generations for r in results]

        run_manager.on_llm_end(LLMResult(generations=generations, llm_output=llm_output))

        return LLMResult(generations=generations, llm_output=llm_output)

    def _generate_with_cache(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        # 9) 检查缓存
        llm_cache = _resolve_cache(cache=self.cache)
        if llm_cache:
            llm_string = self._get_llm_string(stop=stop, **kwargs)
            prompt = dumps(messages)
            cache_val = llm_cache.lookup(prompt, llm_string)
            if cache_val:
                return ChatResult(generations=cache_val)

        # 10) 缓存未命中,调用实际生成
        result = self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)

        # 11) 写入缓存
        if llm_cache:
            llm_cache.update(prompt, llm_string, result.generations)

        return result

    @abstractmethod
    def _generate(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        """子类必须实现的核心生成方法"""
        ...

# 用户代码
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
response = model.invoke("讲一个笑话")

# 调用栈:
# 1. BaseChatModel.invoke
#    ├─ 2. _convert_input: 将str转换为[HumanMessage]
#    ├─ 3. generate_prompt: 包装为PromptValue
#    │    └─ 4. generate: 批量生成(此处只有1个)
#    │         └─ 5. _generate_with_cache: 检查缓存
#    │              ├─ 6. llm_cache.lookup: 查询缓存
#    │              └─ 7. ChatOpenAI._generate: 实际API调用
#    │                   ├─ 8. openai.chat.completions.create: OpenAI API
#    │                   └─ 9. _create_chat_result: 包装为ChatResult
#    └─ 10. 提取generations[0][0].message返回

# 在链中使用
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
chain = prompt | model

result = chain.invoke({"topic": "程序员"})
# prompt.invoke → 生成消息列表 → model.invoke → AIMessage

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Model as BaseChatModel
    participant Cache as LLMCache
    participant CB as CallbackManager
    participant Impl as ChatOpenAI
    participant API as OpenAI API

    User->>Model: invoke("讲一个笑话")
    Model->>Model: _convert_input → [HumanMessage]
    Model->>CB: on_chat_model_start(messages)

    Model->>Cache: lookup(messages, llm_string)
    Cache-->>Model: None (缓存未命中)

    Model->>Impl: _generate(messages)
    Impl->>Impl: 构造API请求参数
    Impl->>API: POST /v1/chat/completions
    Note over API: 模型推理<br/>耗时1-5秒
    API-->>Impl: {choices: [...], usage: {...}}

    Impl->>Impl: _create_chat_result
    Impl-->>Model: ChatResult(generations=[...])

    Model->>Cache: update(messages, llm_string, generations)
    Model->>CB: on_llm_end(result)

    Model-->>User: AIMessage(content="为什么...")

# stream返回Iterator[AIMessageChunk]
class AIMessageChunk(BaseMessageChunk):
    content: str | list  # 增量内容
    tool_call_chunks: list[ToolCallChunk]  # 增量工具调用

class BaseChatModel(BaseLanguageModel):
    def stream(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> Iterator[AIMessageChunk]:
        config = ensure_config(config)
        messages = self._convert_input(input).to_messages()

        # 调用_stream_with_aggregation
        iterator = self._stream_with_aggregation(
            messages,
            stop=stop,
            config=config,
            **kwargs,
        )

        for chunk in iterator:
            yield chunk

    def _stream_with_aggregation(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> Iterator[AIMessageChunk]:
        # 1) 创建回调管理器
        callback_manager = get_callback_manager_for_config(config)
        run_manager = callback_manager.on_chat_model_start(
            dumpd(self),
            [messages],
        )

        # 2) 调用_stream生成器
        generation: ChatGenerationChunk | None = None
        for chunk in self._stream(messages, stop=stop, run_manager=run_manager, **kwargs):
            # 3) 聚合chunk
            if generation is None:
                generation = chunk
            else:
                generation += chunk

            # 4) 触发on_llm_new_token回调
            if run_manager and isinstance(chunk.message.content, str):
                run_manager.on_llm_new_token(
                    chunk.message.content,
                    chunk=chunk,
                )

            # 5) yield消息chunk
            yield chunk.message

        # 6) 结束回调
        if run_manager and generation:
            run_manager.on_llm_end(ChatResult(generations=[generation]))

    def _stream(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        """子类可选实现流式生成

        默认实现:调用_generate并yield整个结果
        """
        result = self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
        yield ChatGenerationChunk(
            message=AIMessageChunk(content=result.generations[0].message.content)
        )

# 流式输出示例
model = ChatOpenAI(model="gpt-4o", streaming=True)

for chunk in model.stream("讲一个长故事"):
    print(chunk.content, end="", flush=True)

# 在Web应用中流式响应
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    def generate():
        for chunk in model.stream(request.messages):
            yield f"data: {json.dumps({'content': chunk.content})}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

sequenceDiagram
    autonumber
    participant User
    participant Model as BaseChatModel
    participant Stream as _stream生成器
    participant API as OpenAI Stream API

    User->>Model: stream("讲故事")
    Model->>Stream: _stream(messages)
    Stream->>API: POST /v1/chat/completions<br/>(stream=True)

    loop 每个token
        API-->>Stream: chunk1: "从"
        Stream->>Stream: on_llm_new_token
        Stream-->>Model: AIMessageChunk("从")
        Model-->>User: yield chunk
        Note over User: 立即显示"从"

        API-->>Stream: chunk2: "前"
        Stream-->>Model: AIMessageChunk("前")
        Model-->>User: yield chunk
        Note over User: 立即显示"前"
    end

    API-->>Stream: [DONE]
    Stream->>Stream: on_llm_end
    Stream-->>Model: 生成器结束

tools: Sequence[dict | type | Callable | BaseTool]  # 工具列表
tool_choice: str | dict | None  # 工具选择策略: "auto" | "any" | "none" | {"name": "tool_name"}
**kwargs: Any  # 额外参数

Runnable[LanguageModelInput, AIMessage]  # 绑定了工具的Runnable

class BaseChatModel(BaseLanguageModel):
    def bind_tools(
        self,
        tools: Sequence[dict | type | Callable | BaseTool],
        *,
        tool_choice: str | dict | None = None,
        **kwargs: Any,
    ) -> Runnable[LanguageModelInput, AIMessage]:
        # 1) 转换工具为标准格式
        formatted_tools = [self._convert_tool(tool) for tool in tools]

        # 2) 构造tools参数(模型特定格式)
        # 对于OpenAI: {"tools": [...], "tool_choice": ...}
        # 对于Anthropic: {"tools": [...]}
        tools_kwargs = self._bind_tools_kwargs(
            formatted_tools,
            tool_choice=tool_choice,
        )

        # 3) 使用bind绑定参数
        return self.bind(**tools_kwargs, **kwargs)

    def _convert_tool(self, tool: dict | type | Callable | BaseTool) -> dict:
        """将工具转换为OpenAI工具格式"""
        if isinstance(tool, dict):
            return tool
        elif isinstance(tool, type) and is_basemodel_subclass(tool):
            # Pydantic模型 → JSON Schema
            return convert_to_openai_tool(tool)
        elif isinstance(tool, BaseTool):
            # LangChain Tool → OpenAI工具
            return convert_to_openai_tool(tool)
        elif callable(tool):
            # 函数 → OpenAI工具
            return convert_to_openai_tool(tool)
        else:
            msg = f"Unsupported tool type: {type(tool)}"
            raise ValueError(msg)

    def _bind_tools_kwargs(
        self,
        tools: list[dict],
        tool_choice: str | dict | None = None,
    ) -> dict:
        """子类重写以适配模型特定格式"""
        # OpenAI格式
        kwargs = {"tools": tools}
        if tool_choice:
            kwargs["tool_choice"] = tool_choice
        return kwargs

from langchain_core.tools import tool

# 定义工具
@tool
def get_weather(location: str) -> str:
    """获取指定地点的天气信息

    Args:
        location: 地点名称,如"北京"
    """
    # 实际实现...
    return f"{location}的天气是晴天,25度"

@tool
def search_web(query: str) -> str:
    """搜索互联网

    Args:
        query: 搜索查询
    """
    # 实际实现...
    return f"搜索结果: {query}"

# 绑定工具到模型
model = ChatOpenAI(model="gpt-4o")
model_with_tools = model.bind_tools([get_weather, search_web])

# 调用
response = model_with_tools.invoke("北京天气怎么样?")
print(response.tool_calls)
# [{'name': 'get_weather', 'args': {'location': '北京'}, 'id': 'call_123'}]

# Agent模式:自动执行工具
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model, [get_weather, search_web])
result = agent.invoke({"messages": [("user", "北京天气怎么样?")]})

sequenceDiagram
    autonumber
    participant User
    participant ModelWithTools as 绑定工具的模型
    participant API as LLM API

    User->>ModelWithTools: invoke("北京天气?")
    ModelWithTools->>API: POST /chat/completions<br/>{messages:[...], tools:[...]}

    Note over API: 模型分析输入<br/>决定调用工具

    API-->>ModelWithTools: AIMessage(<br/>  tool_calls=[{<br/>    name: "get_weather",<br/>    args: {"location": "北京"}<br/>  }]<br/>)

    ModelWithTools-->>User: AIMessage (包含tool_calls)

    Note over User: 用户代码执行工具<br/>并将结果发回

    User->>ModelWithTools: invoke([<br/>  HumanMessage("北京天气?"),<br/>  AIMessage(tool_calls=[...]),<br/>  ToolMessage("晴天25度")<br/>])

    ModelWithTools->>API: POST /chat/completions
    API-->>ModelWithTools: AIMessage("北京今天晴天...")
    ModelWithTools-->>User: 最终答案

schema: type[T] | dict  # Pydantic模型类或JSON Schema
method: Literal["function_calling", "json_mode"]  # 实现方法
include_raw: bool  # 是否包含原始AIMessage
**kwargs: Any  # 额外参数

Runnable[LanguageModelInput, T]  # 输出为schema指定的类型T
# 或 Runnable[LanguageModelInput, dict[str, T | AIMessage]] 当include_raw=True

class BaseChatModel(BaseLanguageModel):
    def with_structured_output(
        self,
        schema: type[T] | dict,
        *,
        method: Literal["function_calling", "json_mode"] = "function_calling",
        include_raw: bool = False,
        **kwargs: Any,
    ) -> Runnable[LanguageModelInput, T]:
        # 1) 根据method选择实现策略
        if method == "function_calling":
            # 使用bind_tools + 解析器
            if isinstance(schema, type):
                # Pydantic模型
                tool_name = schema.__name__
                tool = convert_to_openai_tool(schema)
                llm = self.bind_tools([tool], tool_choice=tool_name, **kwargs)
                output_parser = PydanticToolsParser(tools=[schema], first_tool_only=True)
            else:
                # JSON Schema字典
                llm = self.bind_tools([schema], tool_choice=schema["name"], **kwargs)
                output_parser = JsonOutputKeyToolsParser(key_name=schema["name"], first_tool_only=True)

        elif method == "json_mode":
            # 使用JSON模式 + 提示词约束
            llm = self.bind(response_format={"type": "json_object"}, **kwargs)
            if isinstance(schema, type):
                output_parser = JsonOutputParser(pydantic_object=schema)
            else:
                output_parser = JsonOutputParser()

        else:
            msg = f"Unsupported method: {method}"
            raise ValueError(msg)

        # 2) 构建链
        if include_raw:
            # 返回 {"raw": AIMessage, "parsed": T}
            return RunnableMap({
                "raw": llm,
                "parsed": llm | output_parser,
            })
        else:
            # 只返回解析后的T
            return llm | output_parser

from pydantic import BaseModel, Field

# 定义输出schema
class Joke(BaseModel):
    setup: str = Field(description="笑话的铺垫")
    punchline: str = Field(description="笑点/包袱")
    rating: int = Field(description="有趣程度1-10")

# 创建结构化输出模型
model = ChatOpenAI(model="gpt-4o")
structured_model = model.with_structured_output(Joke)

# 调用,自动解析为Joke对象
joke = structured_model.invoke("讲一个程序员的笑话")
print(f"铺垫: {joke.setup}")
print(f"包袱: {joke.punchline}")
print(f"评分: {joke.rating}")

# 获取原始消息
structured_model_with_raw = model.with_structured_output(Joke, include_raw=True)
result = structured_model_with_raw.invoke("讲一个笑话")
print(result["parsed"])  # Joke对象
print(result["raw"])     # AIMessage

sequenceDiagram
    autonumber
    participant User
    participant Chain as structured_model
    participant LLM as bind_tools(schema)
    participant Parser as PydanticParser

    User->>Chain: invoke("讲笑话")
    Chain->>LLM: invoke(input)
    LLM->>LLM: 生成响应<br/>(tool_calls格式)
    LLM-->>Chain: AIMessage(tool_calls=[{<br/>  name: "Joke",<br/>  args: {<br/>    setup: "...",<br/>    punchline: "...",<br/>    rating: 8<br/>  }<br/>}])

    Chain->>Parser: parse(AIMessage)
    Parser->>Parser: 提取tool_calls[0].args
    Parser->>Parser: 验证Pydantic模型
    Parser-->>Chain: Joke(setup="...", punchline="...", rating=8)

    Chain-->>User: Joke对象