注：以下所提的“文档”，是指LangGraph官方指南（https://langchain-ai.github.io/langgraph/guides/），参考版本是0.6.8。

以下代码的开发环境：

[project]
name = "my-langgraph"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "dotenv>=0.9.9",
    "langchain>=0.3.27",
    "langchain-openai>=0.3.33",
    "langfuse>=3.5.0",
    "langgraph>=0.6.7",
    "langgraph-checkpoint-sqlite>=2.0.11",
    "numpy>=2.3.3",
]

一、文档说明

• 这篇文章属于 LangGraph 在 Agent / 模型集成层面的说明，展示如何在 Agent / 工作流里配置、使用 LLM（聊天模型）与模型选择策略。
• 指南文档在开头就指出：这一系列文档在 LangGraph v1.0（2025 年 10 月） 之后将废弃或重构。也就是说，这里的 API / 模型集成写法可能在未来版本中会有变化。

二、核心内容结构 + 要点

文档围绕以下几个主题组织：

1. 初始化模型（Initialize a model）
2. 直接实例化模型（Instantiate directly）
3. 在 Agent 中使用模型（Use in an agent）
4. 动态模型选择（Dynamic model selection）
5. 高级模型配置（Advanced model configuration），包括禁用流、模型 fallback、速率限制等
6. 自带 / 自定义模型（Bring your own model）

下面依次总结这些部分。

1. 初始化模型（init_chat_model）

• 文档推荐使用 langchain.chat_models.init_chat_model(...) 这个统一入口来初始化聊天模型（LLM）。支持多个后端 / 提供者，如 OpenAI、Anthropic、Azure、Google、AWS Bedrock 等。
• 举例：

from langchain.chat_models import init_chat_model 
llm = init_chat_model("openai:gpt-4.1")

• 这种方式简化了模型初始化、统一了配置路径，也便于使用语言链生态系统的模型支持。文档列出几种常见提供者（OpenAI, Anthropic, Azure, Google, AWS）作为支持示例。

2. 直接实例化模型

• 如果某个模型不在 init_chat_model 支持列表中（或你有自定义模型），也可以直接用该模型类（符合 LangChain / BaseChatModel 接口）来实例化。文档提到例子如 ChatAnthropic。
• 但条件是：这个模型类要实现 LangChain 的 BaseChatModel 接口，而且要支持或兼容 tool calling（如果你希望模型能用工具）。

3. 在 Agent 中使用模型

• LangGraph 提供预构建 Agent（如 create_react_agent）接口，你可以传一个模型名称（字符串）或模型实例给 Agent，Agent 会内部调用 init_chat_model 或基于该实例构造逻辑。
• 例如：

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[...],
)

这样你不必手动 import / 实例化模型。

4. 动态模型选择（Dynamic model selection）

• 有些场景你希望在运行时根据状态 /上下文 /用户配置选择模型，而不是在 Agent 构建时固定模型。文档支持把 model 参数传为一个函数（callable），在调用 Agent 时动态返回模型实例。
• 在使用动态模型选择时，需要注意：若你的 Agent 有 tools，要确保在这个 selector 函数里给模型绑定工具（model.bind_tools([...])）以使工具调用可用。
• 文档举例说明如何依据 runtime.context（静态 context）来选择模型实例。

5. 高级模型配置

文档列出若干可用但非必需的模型 / 模型行为控制方案。这些选项能帮助你在复杂场景里调优或做健壮处理。

6. Bring Your Own Model / 自定义模型

• 如果你使用的是一个非标准 / 自己实现的 LLM（比如自己训练的或者内部私有模型），可以通过两种方式接入：
  1. 让它符合 BaseChatModel 接口：实现所需的方法 /属性，这样它可以无缝作为模型被 Agent / Graph 使用。文档提到这是被推荐的方式。
  2. 自定义流式调用 + StreamWriter：如果你的模型有特别的流式输出方式，可以自己写流式逻辑，把流片段发给 LangGraph 的 writer，集成进 Graph / Agent 的流体系中。文档指出这个路径适用于比较底层 / 高度自定义的场景。

三、补充理解与使用建议

基于上述内容和我自己的实践经验，这里是一些补充理解、建议和潜在注意点。

补充理解

• 模型配置 → Agent / Graph 的关键依赖：在 LangGraph 中，模型是 Agent / 执行流程里的核心“智力”组件，把 prompt → 响应逻辑封装在模型内部或在 wrapper 里。因此模型的初始化方式、可配置性、动态切换能力，对整个系统灵活性影响很大。
• 动态模型切换 是一个很有用的扩展：例如根据任务难度 /成本 /延迟选择更强模型或更轻量模型，适合资源敏感或多用户场景。
• fallback / rate limit 是增强健壮性的好手段，能够在主模型失败或因速率限制被阻断时自动降级、重试或替代。

使用建议 / 注意事项

1. 确认未来版本变动风险
   文档开头明确指出该模块在 v1.0 将被废弃或重构。使用时要留出升级适配空间。
2. 流式与非流式兼容
   如果你使用 disable_streaming=True，就不能用 stream_mode="messages" 那种 token 级输出，要保证你的调用 /消费方式和模型配置一致。
3. 工具调用绑定
   使用动态模型选择时，务必要在 selector 函数里为模型显式绑定工具（model.bind_tools([...])），否则工具调用可能失效。文档对此有明确说明。
4. fallback 顺序与成本权衡
   fallback 模型虽然提供健壮性，但也可能带来延迟 /成本（次优模型请求也消耗资源）。要根据业务场景设计 fallback 优先级与使用条件。
5. Rate limiter 使用
   默认模型可能没有限速，若你在高并发 / 多线程环境下使用同一个模型实例，要考虑用 RateLimiter 控制请求频率，以避免触发 API 限制、超出配额或遭到封禁。
6. 嵌入 / 语义记忆 联合配置
   虽然这篇文档不直接谈 memory，但在一个 Agent + memory 的系统里，模型、memory、context 三者协作很关键。模型 prompt 构造常常要把 memory 检索结果 / context 融入前缀 / system prompt 中。
7. 离线模型 / 内部模型集成
   如果你的模型不在 LangChain 支持范围，可以通过 “Bring your own model” 方式接入。但要注意流式输出 / 错误处理 /工具调用等边界实现。

四、代码例子

import argparse, os, time
from typing import TypedDict, Annotated, Dict, Any, List
from operator import add

from langgraph.runtime import get_runtime
from pydantic import BaseModel

from langgraph.graph import StateGraph, START, END
from langgraph.config import get_stream_writer

# 统一的模型初始化入口（LC 0.3+）
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

"""
要点：
用 workflow（StateGraph） 串 3 个节点：draft → refine → style。
通过 context（静态上下文） 动态选择模型（支持 OpenAI / Anthropic，谁可用就用谁），并配置 fallback 列表。
（可选）开启 token 级流式：如果模型和环境支持，stream_mode=["messages"] 就能边打字边出。
代码自动根据环境变量挑选可用提供商；你只需设置可用的 API Key。
"""

# ========= 1) 上下文（静态）与状态（动态） =========
class Ctx(BaseModel):
    # 层级：fast（便宜/快）| smart（更强/更贵）
    tier: str = "fast"
    # 可选：优先使用的提供商序列（会自动挑可用的）
    # 例如 ["openai","anthropic"] 或反过来
    providers: List[str] = ["openai", "anthropic"]
    # 生成温度
    temperature: float = 0.2
    # 关闭 token 级流式（如需一次性出结果）
    disable_streaming: bool = False
    # 目标写作风格
    tone: str = "concise"

class FlowState(TypedDict, total=False):
    topic: str
    draft: str
    refined: str
    styled: str
    log: Annotated[List[str], add]

# ========= 2) 动态模型选择 + Fallback =========
def _candidate_models(tier: str, provider: str) -> List[str]:
    """
    返回某 provider 在指定 tier 下的候选模型（按优先级）。
    你可按需要替换成你的模型清单。
    """
    if provider == "openai":
        if tier == "smart":
            return ["openai:gpt-4.1-mini", "openai:gpt-4o-mini"]
        else:  # fast
            return ["openai:gpt-4o-mini", "openai:gpt-4o-mini-2024-08-06"]
    if provider == "anthropic":
        if tier == "smart":
            return ["anthropic:claude-3-7-sonnet-latest", "anthropic:claude-3-5-sonnet-latest"]
        else:
            return ["anthropic:claude-3-5-haiku-latest", "anthropic:claude-3-5-sonnet-latest"]
    return []

def build_llm(runtime_ctx: Ctx):
    """
    - 遍历 providers，按 tier 拿到候选模型清单
    - 用第一个“可正常初始化”的模型作为主模型
    - 其余成功初始化的作为 fallbacks（自动降级）
    """
    streaming = not runtime_ctx.disable_streaming
    temperature = runtime_ctx.temperature

    primary = None
    fallbacks = []

    # 收集候选（按 providers 顺序）
    all_candidates: List[str] = []
    for prov in runtime_ctx.providers:
        all_candidates.extend(_candidate_models(runtime_ctx.tier, prov))

    # 逐个尝试初始化（第一个成功的是主模型，后续成功的是备用）
    for mid in all_candidates:
        try:
            m = init_chat_model(mid, temperature=temperature, streaming=streaming)
            if primary is None:
                primary = m
            else:
                fallbacks.append(m)
        except Exception:
            # 模型不可用（无对应 API Key/未安装 provider/名称不对） → 跳过
            continue

    if primary is None:
        raise RuntimeError(
            "No LLM available. Please set an API key (OPENAI_API_KEY or ANTHROPIC_API_KEY) "
            "and/or adjust providers/models."
        )

    if fallbacks:
        primary = primary.with_fallbacks(fallbacks)
    return primary

# ========= 3) 三个工作流节点 =========
def node_draft(state: FlowState) -> Dict[str, Any]:
    w = get_stream_writer()
    rt = get_runtime(Ctx)
    llm = build_llm(rt.context)

    topic = state.get("topic") or "Modern software release notes"
    sys = "You summarize topics into a clear outline. Keep it tight."
    user = f"Draft a short outline (5 bullets max) about: {topic}"

    w({"node": "draft", "model_tier": rt.context.tier, "providers": rt.context.providers})
    resp = llm.invoke([SystemMessage(content=sys), HumanMessage(content=user)])
    draft = resp.content.strip()
    return {"draft": draft, "log": [f"drafted with tier={rt.context.tier}"]}

def node_refine(state: FlowState) -> Dict[str, Any]:
    w = get_stream_writer()
    rt = get_runtime(Ctx)
    llm = build_llm(rt.context)

    draft = state.get("draft", "")
    sys = "You are an expert editor. Improve clarity and add one concrete example."
    user = f"Refine the outline below, keep it compact:\n\n{draft}"

    w({"node": "refine", "info": "editing"})
    resp = llm.invoke([SystemMessage(content=sys), HumanMessage(content=user)])
    refined = resp.content.strip()
    return {"refined": refined, "log": ["refined"]}

def node_style(state: FlowState) -> Dict[str, Any]:
    w = get_stream_writer()
    rt = get_runtime(Ctx)
    llm = build_llm(rt.context)

    refined = state.get("refined", "")
    sys = f"You are a stylist. Rewrite to be {rt.context.tone}. Preserve info."
    user = refined

    w({"node": "style", "tone": rt.context.tone})
    resp = llm.invoke([SystemMessage(content=sys), HumanMessage(content=user)])
    styled = resp.content.strip()
    return {"styled": styled, "log": [f"styled({rt.context.tone})"]}

# ========= 4) 组装工作流 =========
from langgraph.graph import StateGraph
def build_app():
    g = StateGraph(FlowState, context_schema=Ctx)
    g.add_node("draft", node_draft)
    g.add_node("refine", node_refine)
    g.add_node("style", node_style)

    g.add_edge(START, "draft")
    g.add_edge("draft", "refine")
    g.add_edge("refine", "style")
    g.add_edge("style", END)
    return g.compile()

"""
怎么用
# 走“快模型”，优先 OpenAI，再 Anthropic；保持流式开启（若支持）
python workflow_models_demo.py --tier fast --providers openai,anthropic --topic "Shipping a mobile app update"

# 走“强模型”，优先 Anthropic，再 OpenAI；关闭流式（一次性出）
python workflow_models_demo.py --tier smart --providers anthropic,openai --topic "Zero-downtime DB migrations" --no-stream

# 调整风格
python workflow_models_demo.py --tone executive

你能学到的点
- 工作流里动态选模型：根据 context 做路由，不依赖 Agent。
- Fallback：主模型挂了自动降级到备选，提升健壮性。
- 流式或非流式：按 context 开关，且前端消费通过 stream_mode="messages"（模型需支持）。
- 强类型 Context：context_schema=Ctx，在节点里 get_runtime(Ctx) 读取统一配置。
"""

# ========= 5) CLI =========
def main(topic, tier, providers, tone, no_stream):
    ctx = Ctx(
        tier=tier,
        providers=[s.strip() for s in providers.split(",") if s.strip()],
        tone=tone,
        disable_streaming=bool(no_stream),
    )

    app = build_app()

    print(f"\n=== RUN (tier={ctx.tier}, providers={ctx.providers}, tone={ctx.tone}, streaming={not ctx.disable_streaming}) ===")
    # 如果你的模型支持 token 流式，开启 "messages" 就能看到逐 token 输出
    for mode, chunk in app.stream(
        {"topic": topic},
        context=ctx.model_dump(),
        stream_mode=["updates","custom","messages"],  # "messages" 为 token 级（若模型支持）
    ):
        if mode == "custom":
            print("🟣 [custom ]", chunk)
        elif mode == "updates":
            print("🔸 [updates]", chunk)
        elif mode == "messages":
            token, meta = chunk
            print(token, end="", flush=True)

    # 输出最终结果
    final = app.invoke({}, context=ctx.model_dump())
    print("\n\n--- FINAL OUTPUT ---")
    print(final["styled"])
    print("\n--- LOG ---")
    print(" • " + "\n • ".join(final.get("log", [])))

def run_console():
    parser = argparse.ArgumentParser(description="Workflow + Dynamic Model Selection + Fallback Demo")
    parser.add_argument("--topic", default="How to introduce feature flags safely")
    parser.add_argument("--tier", default="fast", choices=["fast", "smart"])
    parser.add_argument("--providers", default="openai,anthropic",
                        help="comma list, e.g. openai,anthropic or anthropic,openai")
    parser.add_argument("--tone", default="concise", help="final style, e.g. concise, executive, friendly")
    parser.add_argument("--no-stream", action="store_true", help="disable token streaming at the model level")
    args = parser.parse_args()
    main(args.topic, args.tier, args.providers, args.tone, args.no_stream)

def models_example():
    # --tier fast --providers openai,anthropic --topic "Shipping a mobile app update"
    main(topic="Shipping a mobile app update", tier="fast", providers="openai,anthropic", tone="concise", no_stream=True)