1
2
3
4
5

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent

model = init_chat_model("google_genai:gemini-3.5-flash", timeout=60, max_retries=6)
agent = create_agent(model=model, tools=[])

1
2

response = model.invoke("请用一句话解释 LangChain 的模型抽象。")
print(response.text)

1
2

for chunk in model.stream("请解释模型流式输出的价值。"):
    print(chunk.text, end="", flush=True)

1
2
3
4

responses = model.batch(
    ["解释工具调用。", "解释结构化输出。"],
    config={"max_concurrency": 2},
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from langchain.tools import tool

@tool
def get_weather(location: str) -> str:
    """查询指定城市的天气。"""
    return f"{location} 天气晴朗。"

model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("查询北京天气。")
print(response.tool_calls)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from pydantic import BaseModel, Field

class Movie(BaseModel):
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="十分制评分")

model_with_structure = model.with_structured_output(
    Movie,
    method="json_schema",
    include_raw=True,
)

response = model_with_structure.invoke("请给出电影《盗梦空间》的基本信息。")
print(response["parsed"])
print(response["raw"])
print(response["parsing_error"])

1
2

tool = {"type": "web_search"}
model_with_tools = model.bind_tools([tool])

1
2
3
4
5
6

model = init_chat_model(
    model="MODEL_NAME",
    model_provider="openai",
    base_url="https://your-vllm-server.example.com/v1",
    api_key="YOUR_API_KEY",
)

1
2
3
4
5
6

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-5.5",
    openai_proxy="http://proxy.example.com:8080",
)

1
2
3
4
5
6

from langchain_openrouter import ChatOpenRouter

model = ChatOpenRouter(
    model="auto",
    api_key=os.environ["OPENROUTER_API_KEY"],
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="gpt-5.5",
    model_provider="openai",
).bind(logprobs=True)

response = model.invoke("请用一句话解释 LangChain 的模型抽象。")

# token 级别概率信息
print(response.response_metadata["logprobs"])

# token 用量信息
print(response.usage_metadata)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from langchain.chat_models import init_chat_model
from langchain_core.callbacks import get_usage_metadata_callback

model_1 = init_chat_model(model="gpt-5.4-mini")
model_2 = init_chat_model(model="claude-haiku-4-5-20251001")

with get_usage_metadata_callback() as cb:
    model_1.invoke("请用一句话解释 token。")
    model_2.invoke("请用一句话解释模型成本统计。")
    print(cb.usage_metadata)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-5.5", model_provider="openai")

responses = model.batch(
    [
        "请解释 invoke()。",
        "请解释 stream()。",
        "请解释 batch()。",
    ],
    config={
        "run_name": "模型 API 批处理解释",
        "tags": ["models", "batch"],
        "metadata": {"user_id": "user-123"},
        "max_concurrency": 2,
    },
)

for response in responses:
    print(response.text)

1
2
3
4
5
6

configurable_model = init_chat_model(temperature=0)

configurable_model.invoke(
    "请介绍你自己。",
    config={"configurable": {"model": "gpt-5-nano"}},
)

  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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Author : nanzet
# Description : 演示 LangChain ChatModel 的初始化、调用、流式输出、工具调用、结构化输出和 token 用量统计
# requirements : pip install -U langchain-google-genai pydantic typing_extensions

import json
import os
import sys
from typing import Any

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.callbacks import UsageMetadataCallbackHandler
from pydantic import BaseModel, Field

MODEL_NAME = "google_genai:gemini-3.5-flash"


class Movie(BaseModel):
    """电影结构化信息。"""

    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="十分制评分")


@tool
def get_weather(location: str) -> str:
    """查询指定地点的天气。"""
    return f"{location} 当前天气晴朗，适合外出。"


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"缺少环境变量：{name}。请设置后重新运行脚本。")
    return value


def text_of(message: Any) -> str:
    text = getattr(message, "text", None)
    if isinstance(text, str) and text:
        return text

    content = getattr(message, "content", "")
    if isinstance(content, str):
        return content

    if isinstance(content, list):
        parts: list[str] = []
        for block in content:
            if isinstance(block, dict) and block.get("type") == "text":
                parts.append(str(block.get("text", "")))
        return "\n".join(part for part in parts if part)

    return str(message)


def print_structured_response(response: Any) -> None:
    if hasattr(response, "model_dump_json"):
        print(response.model_dump_json(indent=2))
    elif isinstance(response, (dict, list)):
        print(json.dumps(response, ensure_ascii=False, indent=2))
    else:
        print(str(response))


def run_basic_invocation(model: Any) -> None:
    print("基础调用结果：")
    response = model.invoke("请用两句话解释 LangChain 的模型抽象。")
    print(text_of(response))


def run_streaming(model: Any) -> None:
    print("\n流式输出结果：")
    for chunk in model.stream("请用三点说明为什么模型流式输出能改善用户体验。"):
        print(text_of(chunk), end="", flush=True)
    print()


def run_tool_calling(model: Any) -> None:
    print("\n工具调用结果：")
    available_tools = {"get_weather": get_weather}
    model_with_tools = model.bind_tools(list(available_tools.values()))
    messages: list[Any] = [
        {"role": "user", "content": "请查询北京和上海的天气，并给出简短建议。"}
    ]

    ai_message = model_with_tools.invoke(messages)
    messages.append(ai_message)

    if not ai_message.tool_calls:
        print("模型没有请求调用工具，直接输出：")
        print(text_of(ai_message))
        return

    for tool_call in ai_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        print(f"准备执行工具：{tool_name}，参数：{tool_args}")

        selected_tool = available_tools.get(tool_name)
        if selected_tool is None:
            print(f"模型请求了未注册工具：{tool_name}，已跳过该工具调用。")
            continue

        tool_result = selected_tool.invoke(tool_call)
        messages.append(tool_result)

    final_response = model_with_tools.invoke(messages)
    print(text_of(final_response))


def run_structured_output(model: Any) -> None:
    print("\n结构化输出结果：")
    model_with_structure = model.with_structured_output(Movie)
    response = model_with_structure.invoke("请给出电影《盗梦空间》的基本信息。")
    print_structured_response(response)


def run_usage_tracking(model: Any) -> None:
    print("\nToken 用量统计：")
    callback = UsageMetadataCallbackHandler()
    response = model.invoke(
        "请用一句话说明 token 用量统计在生产环境中的价值。",
        config={
            "run_name": "模型用量统计示例",
            "tags": ["models", "usage"],
            "metadata": {"user_id": "demo-user"},
            "callbacks": [callback],
        },
    )
    print(text_of(response))
    print(callback.usage_metadata)


def main() -> int:
    try:
        require_env("GOOGLE_API_KEY")

        model = init_chat_model(
            MODEL_NAME,
            temperature=0,
            timeout=60,
            max_tokens=2000,
            max_retries=6,
        )

        run_basic_invocation(model)
        run_streaming(model)
        run_tool_calling(model)
        run_structured_output(model)
        run_usage_tracking(model)
        return 0

    except ImportError as exc:
        print(f"依赖导入失败：{exc}")
        print(
            "请执行：pip install -U langchain-google-genai pydantic typing_extensions"
        )
        return 1
    except Exception as exc:
        print(f"脚本执行失败：{exc}")
        return 1


if __name__ == "__main__":
    sys.exit(main())

 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

基础调用结果：
LangChain的模型抽象通过统一的接口，将不同厂商的大模型标准化为“语言模型（LLMs）”和“聊天模型（Chat Models）”两大类。

这种设计屏蔽了底层API的细节差异，使开发者无需修改核心代码，即可在不同的模型之间进行无缝切换与组合。

流式输出结果：
模型流式输出（Streaming Output，即字词逐个或逐块实时呈现）是现代大语言模型应用（如 ChatGPT）的标配。它能显著改善用户体验，原因主要体现在以下三点：

### 1. 显著降低“首字延迟”，消除用户的等待焦虑
* **原理解析**：大模型生成完整回答通常需要数秒甚至数十秒。如果采用“非流式输出”（一次性返回全部结果），用户在等待期间只能面对加载动画（如 Loading 菊花图），容易产生系统卡死或响应缓慢的挫败感。
* **体验提升**：流式输出让**首字响应时间（Time to First Token, TTFT）**缩短至几百毫秒。用户几乎在按下发送键的瞬间就能看到模型开始“说话”，这种即时反馈极大地缓解了等待焦虑，让交互显得流畅且高效。

### 2. 符合人类阅读习惯，提供更自然的“渐进式”认知体验
* **原理解析**：人类的阅读和信息处理是线性的。如果瞬间弹出一大段文字（“信息轰炸”），用户需要花时间重新定位阅读起点，容易产生视觉疲劳和认知过载。
* **体验提升**：流式输出模仿了人类说话或打字的自然节奏。用户可以跟随文字生成的进度**边出边读**。这种渐进式的信息呈现方式不仅更具拟人感（像在与真人对话），还能让用户的大脑同步消化信息，提高阅读的舒适度和理解效率。

### 3. 支持实时反馈与即时干预，提高任务处理效率
* **原理解析**：在长文本生成或复杂任务中，模型有时会“幻觉”（胡说八道）或偏离用户的真实意图。
* **体验提升**：通过流式输出，用户可以在模型刚开始生成的几秒内，就判断出方向是否正确。如果发现模型理解有误，用户可以**立即点击“停止生成”**并修改 Prompt（提示词），而不需要傻傻等待模型吐完几百个错字。这种“过程可控性”极大地节省了用户的时间，提升了人机协作的整体效率。

工具调用结果：
准备执行工具：get_weather，参数：{'location': '北京'}
准备执行工具：get_weather，参数：{'location': '上海'}
北京和上海目前的天气都非常好：

*   **北京**：当前天气晴朗，非常适合外出。
*   **上海**：当前天气同样晴朗，适合出行。

**简短建议**：两地今天都是晴空万里，非常适合进行户外活动或晾晒衣物。不过紫外线可能较强，外出时请注意防晒和补水。

结构化输出结果：
{
  "title": "盗梦空间",
  "year": 2010,
  "director": "克里斯托弗·诺兰",
  "rating": 9.4
}

Token 用量统计：
Token 用量统计是生产环境中实现**精细化成本控制**、**精准商业化计费**以及**系统容量与异常监控**的核心数据基石。
{'gemini-3.5-flash': {'input_tokens': 16, 'output_tokens': 712, 'total_tokens': 728, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 676}}}

 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
90
91
92
93
94
95
96
97
98
99

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Author : nanzet
# Description : 演示 LangChain 可配置模型、批处理并发控制和速率限制器
# requirements : pip install -U langchain langchain-deepseek

import os
import sys
from typing import Any

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"缺少环境变量：{name}。请设置后重新运行脚本。")
    return value


def text_of(message: Any) -> str:
    text = getattr(message, "text", None)
    if isinstance(text, str) and text:
        return text
    content = getattr(message, "content", "")
    return content if isinstance(content, str) else str(content)


def main() -> int:
    try:
        require_env("DEEPSEEK_API_KEY")

        rate_limiter = InMemoryRateLimiter(
            requests_per_second=0.2,
            check_every_n_seconds=0.1,
            max_bucket_size=2,
        )

        configurable_model = init_chat_model(
            temperature=0,
            timeout=60,
            max_tokens=3000,
            rate_limiter=rate_limiter,
            configurable_fields=(
                "model",
                "model_provider",
                "temperature",
                "max_tokens",
            ),
        )

        response = configurable_model.invoke(
            "请用一句话说明运行时可配置模型的价值。",
            config={
                "configurable": {
                    "model": "deepseek-v4-pro",
                    "model_provider": "deepseek",
                }
            },
        )
        print("可配置模型调用结果：")
        print(text_of(response))

        questions = [
            "请解释模型批处理的适用场景。",
            "请解释 max_concurrency 的作用。",
            "请解释 rate_limiter 的作用。",
        ]

        responses = configurable_model.batch(
            questions,
            config={
                "max_concurrency": 2,
                "configurable": {
                    "model": "deepseek-v4-pro",
                    "model_provider": "deepseek",
                },
            },
        )

        print("\n批处理结果：")
        for index, item in enumerate(responses, start=1):
            print(f"{index}. {text_of(item)}")

        return 0

    except ImportError as exc:
        print(f"依赖导入失败：{exc}")
        print("请执行： pip install -U langchain langchain-deepseek")
        return 1
    except Exception as exc:
        print(f"脚本执行失败：{exc}")
        return 1


if __name__ == "__main__":
    sys.exit(main())

  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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

可配置模型调用结果：
运行时可配置模型的价值在于它使系统能够在不停机、不重新部署的情况下动态调整行为，从而实时响应业务需求变化，显著提升敏捷性与连续性。

批处理结果：
1. 模型批处理（Model Batching）是指将多个输入样本合并成一个批次（batch），一次性送入模型进行计算，而非逐个样本串行处理。它在深度学习推理和训练中被广泛使用，其适用场景主要取决于性能收益、延迟要求、硬件特性及数据特性等因素。

以下是模型批处理的核心适用场景和分析：

### 1. 高吞吐量优先的离线处理场景
这是批处理最典型、收益最大的场景。
- **场景举例**：
  - 大规模数据标注或特征提取
  - 离线批量推理（如每日定时对千万级图片打标、文本分类、向量化）
  - 模型训练过程（训练本身就依赖于大批次梯度下降）
- **为何适用**：
  - 不要求低延迟，更关注单位时间内处理的总样本数（吞吐量）。
  - 通过增加批次大小，可以充分利用 GPU 的并行计算单元，提高计算密度，将数据传输和核函数启动的开销均摊到更多样本上。
  - 离线任务通常可以将大量数据预先整理好，天然形成批次。

### 2. GPU/TPU 等高度并行硬件的利用
批处理的最大驱动力来自硬件的并行架构。
- **场景举例**：
  - 任何在 GPU 上运行的卷积神经网络（CNN）或 Transformer 模型推理。
  - 在 CPU 上利用 SIMD（单指令多数据流）指令集时，批次处理也能带来大幅加速（尤其在启用供应商库如 Intel OpenVINO、ONNX Runtime 优化时）。
- **为何适用**：
  - GPU 有成千上万个计算核心，单条样本无法填满算力，会导致计算单元大量空闲。
  - 批处理可以将矩阵乘法等操作从“矩阵-向量”运算转变为“矩阵-矩阵”运算，后者的算术强度（计算/访存比）更高，能更好地隐藏内存访问延迟，达到接近硬件的峰值算力。

### 3. 最大化内存带宽利用率
- **场景举例**：
  - 模型权重加载是瓶颈的轻量级模型（如 MobileNet 在小设备上）。
  - 推理过程中需要频繁从显存/内存读取相同权重与多条输入进行计算。
- **为何适用**：
  - 当处理单个样本时，模型权重必须从内存中读取一次。如果批次中有  N  个样本，权重只需要读取一次，却可以重复用于  N  次计算。
  - 这种方式将内存读取开销均摊，让访存瓶颈型模型变为计算瓶颈型，从而提升吞吐。

### 4. 可容忍可控延迟的在线服务
在线服务通常要求低延迟，但某些场景允许适度增加延迟以换取更高的资源利用率和更低的单次推理成本。
- **场景举例**：
  - 异步推理服务：用户提交任务后不立即等待结果，而是通过回调或轮询获取。
  - 推荐系统的候选集打分阶段：一次请求可能需要对上百个候选 item 打分，天然形成一个批次。
  - 智能客服中同时输入多条相似问法进行意图识别。
- **为何适用**：
  - 可以设置一个极短的时间窗口（如 5–20 毫秒）来攒批：将窗口内到达的请求合并成一个动态批次，一次性推理后拆分结果返回。
  - 这种方法牺牲了微秒到毫秒级的尾部延迟，但可以在保证系统整体响应在可接受范围（如 50ms）内的情况下，数倍提升服务实例的吞吐和硬件利用率。
  - 典型技术：TensorFlow Serving、Triton Inference Server 的动态批处理功能。

### 5. 序列长度或输入形状一致的场景（利于静态批处理）
- **场景举例**：
  - 图像分类、目标检测（经过预处理 resize 后尺寸固定）。
  - 文本分类（通过 padding 对齐到相同长度）。
  - 嵌入向量查找。
- **为何适用**：
  - 如果不需复杂的可变长度处理或掩码，直接拼接成规整张量效率最高。
  - 静态形状可以充分利用计算图的优化和 XLA 等编译加速。
  - 即使长度不一，通过填充（padding）和掩码（mask）仍然可以组成批次，是为动态批处理，在 NLP 模型中极为普遍。

### 6. 模型计算量较大，但并非极端巨大的情况
- **场景举例**：
  - ResNet-50、BERT-base、Stable Diffusion 等模型。
- **为何适用**：
  - 如果模型计算量极小（如简单的逻辑回归），CPU 上可能单条推理就很快，批次会引入组包和解包开销，反而得不偿失。
  - 如果模型极大（如 175B LLM），单个 GPU 显存可能仅能容纳 1 条请求（batch size=1），此时无法进行传统批处理，需转而使用连续批处理技术（如 vLLM 的 continuous batching）来调度注意力机制中的 KV cache，提升吞吐，但这属于更高级的批处理变体。
- **不适用的反例**：
  - 在极度实时、要求逐样本响应的流式场景（如实时语音对话），且硬件资源足够，可以接受单样本推理。
  - 某些边缘端设备，内存极度受限，无法容纳大于 1 的 batch size。

### 总结：决定是否使用批处理的关键决策因子
如果你面对一个具体任务，可以通过以下问题判断：
1. **对延迟的要求是毫秒级实时还是秒级异步？** → 异步或可容忍延迟则积极使用。
2. **硬件是 GPU 还是纯 CPU？** → GPU 上收益巨大；CPU 上需测试，小模型和高并发时也有明显收益。
3. **数据是否可以轻易积累成批？** → 离线全量或攒批窗口内请求量足够。
4. **模型计算量是否足够大？** → 足够大才值得，极简模型受益小。
5. **是否有内存/显存承载更大 batch？** → 需要确保不引发 OOM，并可配合梯度累积（训练）或 offload 技术。

批处理是一项用延迟换吞吐、用空间换时间的核心技术，在大多数非严格实时的深度学习应用中都是适用的，甚至已成为标准操作模式。
2. `max_concurrency` 是一个常见的配置参数，中文可译为“最大并发数”，它的核心作用是**限制同时执行的操作（如请求、任务、线程、协程）数量**。无论在后端服务、异步编程、分布式任务队列还是网络请求中，这个参数都用于控制资源消耗、保护系统稳定性并优化吞吐量。

在不同的技术场景中，它的具体含义和效果如下：

---

### 1. 异步与并发编程（如 `asyncio.Semaphore`、`ThreadPoolExecutor`）
- **限制协程/线程同时运行的数量**  
  当你使用 `asyncio` 发起大量网络请求时，如果同时发起 1000 个请求，可能会耗尽带宽、导致目标服务器限流或本地端口耗尽。通过 `max_concurrency`（常通过 `asyncio.Semaphore(max_concurrency)` 实现），可以保证**任意时刻只有 N 个协程在同时执行**，其余协程排队等待。
- **效果**：平滑负载、避免触发远端限流、降低内存与连接数峰值。

### 2. HTTP 客户端（如 `aiohttp`、`httpx`、爬虫框架）
- **控制对同一域名的并发连接数**  
  例如 `aiohttp` 的 `TCPConnector` 有 `limit` 参数（等同于 `max_concurrency`），它限制同时打开的连接数。  
  **作用**：遵守目标网站的爬虫礼仪、防止被判定为攻击、减少超时和错误重试。

### 3. API 与微服务网关（如 FastAPI 的后台任务、自定义并发控制）
- **保护下游服务**  
  当你的服务需要调用第三方 API，且该 API 有明确的 QPS/并发数限制（如“同一账号最多 5 个并发请求”）时，设置 `max_concurrency=5` 可以确保不会超出限制，避免被拒绝服务。
- **后端压力控制**  
  在消费消息队列或者后台批量任务时，限制并发数可以防止 CPU 满载、数据库连接池耗尽。

### 4. 分布式任务队列（如 Celery）
- **工作进程/协程的并发粒度**  
  Celery 的 `--concurrency` 或 `worker_concurrency` 决定了工作节点**同时能处理多少个任务**。  
  - 如果设为 1，任务严格串行。
  - 如果设为 10，节点同时拉取并执行最多 10 个任务。
  **作用**：平衡任务吞吐量与系统资源（CPU 核数、内存、数据库连接）。

### 5. 大数据与流处理（如 Flink、Spark Streaming）
- **算子并行度与背压控制**  
  虽然常称为 `parallelism`，但有时也会出现 `maxConcurrency` 来限制某个算子的最大并发实例数，**防止下游被冲垮**。

---

### 为什么需要 `max_concurrency`？常见动机总结：
| 如果没有限制 | 设置合适 max_concurrency 后 |
|-------------|---------------------------|
| 瞬间发起海量请求，打垮目标服务 | 请求平缓发送，下游稳定 |
| 本地 CPU/内存/文件描述符被耗尽 | 资源占用平稳可控 |
| 任务全部抢到资源但都超时失败 | 成功率上升，总耗时可能更短 |
| 数据库连接池耗尽，所有操作报错 | 排队等待连接，系统仍有响应 |

### 如何选择合理的值？
- **I/O 密集型任务**（网络请求、磁盘读写）：可以设得较高（如 50～200），受限于带宽、文件描述符和目标服务承受能力。
- **CPU 密集型任务**：建议设为 CPU 核心数（或稍多），避免上下文切换开销。
- **受下游限制**：严格按照下游文档或实测结果（如 API 允许 10 QPS，并发约 2～5）。

一句话总结：**`max_concurrency` 是限制“并行工作单元”数量的闸门，用来在吞吐量与稳定性之间取得平衡。**
3. 限流器（rate limiter）是一种用于控制系统处理请求速率的机制。它的核心作用是**防止系统或服务因短时间内接收到过多请求而过载，从而保障稳定性、可用性和公平性**。

可以从以下几个层面来理解它的作用：

### 1. 保护系统资源，防止过载
无论是 Web 服务器、数据库还是微服务，其处理能力（CPU、内存、I/O、连接数）都是有上限的。如果流量瞬间暴增（如突发新闻、秒杀活动或恶意攻击），系统可能因资源耗尽而响应变慢、崩溃或级联故障。
- **作用示例**：限制每个用户每分钟最多请求 100 次 API。当超过限制时，直接返回 `429 Too Many Requests`，避免后台服务被冲垮。

### 2. 防御恶意攻击和滥用
公共 API 或登录接口常常是恶意行为的目标，例如暴力破解密码、DDoS 攻击、爬虫抓取数据或刷单。
- **作用示例**：限制单个 IP 对登录接口的尝试次数（如 5 分钟内最多 5 次），可有效阻止暴力破解。
- **防止数据爬取**：限制单个账户或 IP 每天可调用的数据导出次数，防止批量窃取信息。

### 3. 保障服务质量与公平性（QoS）
在多租户系统或平台型产品中，某些用户可能会占用过多资源，影响其他用户的正常使用。限流器可以确保资源被公平分配。
- **作用示例**：免费 API 用户每秒 10 次请求，付费用户每秒 1000 次请求。这既实现了商业化隔离，又保证了基础用户体验。
- **隔离故障**：当下游某个服务变慢时，可以对调用该下游的请求进行限流，防止线程池被耗尽，这是一种熔断降级的补充手段（常与熔断器结合使用）。

### 4. 控制成本与第三方依赖
当系统依赖付费的外部服务（如短信发送、人脸识别、云存储接口）时，这些服务通常本身也有调用频率限制或按量计费。本地限流可以避免意外的高额账单。
- **作用示例**：应用内限制短信验证码的发送频率（如同号码 60 秒内只能发送一次，每天上限 5 条），既符合服务商要求，也控制了成本。

### 5. 平滑流量，削峰填谷
当流量呈脉冲式到达时，限流器可以结合队列，将请求以较平稳的速率释放给后端处理，避免后端压力剧烈波动。
- **作用示例**：消息队列消费端使用限流器，匀速拉取并处理消息，防止数据库写入压力瞬间飙升。

---

### 常见限流算法与位置
理解作用后，通常会选择以下算法来实现：

- **固定窗口**：简单计数，但边界处可能产生流量突刺。
- **滑动窗口**：更精细地平滑边界问题，避免突发。
- **漏桶（Leaky Bucket）**：严格平滑输出速率，强制请求以恒定速度被处理，超出则丢弃/排队。
- **令牌桶（Token Bucket）**：允许一定程度的突发流量，同时限制平均速率，是最常用的算法（如 Guava RateLimiter）。
- **分布式限流**：在集群环境中，通常需要借助 Redis + Lua 脚本或专门的网关组件（如 Sentinel、Kong）实现全局统一限流。

### 实施层面
限流器可以部署在多个层级：
- **网关层**：对进入系统的所有流量进行全局限流。
- **应用层/中间件**：对特定服务或接口进行细粒度限流。
- **基础设施层**：通过负载均衡器或容器编排的 ingress 控制器实现。

总而言之，**rate_limiter 是系统稳定性的第一道防护网**，它通过“拒绝一部分请求”来确保“大部分（甚至全部）系统”能够持续正常运行，是构建高可用分布式系统的必备组件。

  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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Author : nanzet
# Description : 演示在 Agent 中使用动态模型路由和结构化输出
# requirements : pip install -U langchain-google-genai pydantic

import os
import sys

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field

BASIC_MODEL_NAME = "gemini-2.5-flash"
ADVANCED_MODEL_NAME = "gemini-3.5-flash"


class ContentAnalysis(BaseModel):
    """内容分析结果。"""

    sentiment: str = Field(description="情感倾向，例如积极、消极或中性")
    key_entities: list[str] = Field(description="关键实体列表")
    summary: str = Field(description="简短中文摘要")


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"缺少环境变量：{name}。请设置后重新运行脚本。")
    return value


def main() -> int:
    try:
        require_env("GOOGLE_API_KEY")

        basic_model = init_chat_model(
            BASIC_MODEL_NAME, model_provider="google_genai", temperature=0
        )
        advanced_model = init_chat_model(
            ADVANCED_MODEL_NAME, model_provider="google_genai", temperature=0
        )

        @wrap_model_call
        def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
            """根据消息数量选择模型。"""
            message_count = len(request.state["messages"])
            if message_count > 3:
                selected_model = advanced_model
                selected_model_name = ADVANCED_MODEL_NAME
            else:
                selected_model = basic_model
                selected_model_name = BASIC_MODEL_NAME

            print(
                f"动态路由：当前 messages 数量={message_count}，"
                f"选择模型={selected_model_name}"
            )
            return handler(request.override(model=selected_model))

        agent = create_agent(
            model=basic_model,
            tools=[],
            middleware=[dynamic_model_selection],
            response_format=ContentAnalysis,
            system_prompt="你是专业内容分析助手，请严格按照结构化 schema 输出中文结果。",
        )

        def run_analysis_case(title: str, messages: list[dict[str, str]]) -> None:
            print(f"\n{title}")
            result = agent.invoke({"messages": messages})
            structured_response = result["structured_response"]
            print("结构化分析结果：")
            print(structured_response.model_dump_json(indent=2))

        run_analysis_case(
            "场景一：短上下文，预期使用基础模型",
            [
                {
                    "role": "user",
                    "content": "这款旗舰手机续航优秀，但夜景算法仍需优化。请给出结构化分析。",
                }
            ],
        )

        run_analysis_case(
            "场景二：长上下文，预期使用高级模型",
            [
                {"role": "user", "content": "我们正在评估一款旗舰手机。"},
                {"role": "assistant", "content": "请继续提供评价信息。"},
                {"role": "user", "content": "它的屏幕色彩准确，性能稳定。"},
                {"role": "assistant", "content": "已记录屏幕和性能信息。"},
                {
                    "role": "user",
                    "content": "但夜景算法偏慢，机身发热明显。请给出结构化分析。",
                },
            ],
        )
        return 0

    except ImportError as exc:
        print(f"依赖导入失败：{exc}")
        print("请执行：pip install -U langchain-google-genai pydantic")
        return 1
    except Exception as exc:
        print(f"脚本执行失败：{exc}")
        return 1


if __name__ == "__main__":
    sys.exit(main())

 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

场景一：短上下文，预期使用基础模型
动态路由：当前 messages 数量=1，选择模型=gemini-2.5-flash
结构化分析结果：
{
  "sentiment": "中性",
  "key_entities": [
    "旗舰手机",
    "续航",
    "夜景算法"
  ],
  "summary": "这款旗舰手机电池续航表现出色，但夜景拍摄的算法有待改进。"
}

场景二：长上下文，预期使用高级模型
动态路由：当前 messages 数量=5，选择模型=gemini-3.5-flash
结构化分析结果：
{
  "sentiment": "中性",
  "key_entities": [
    "旗舰手机",
    "屏幕",
    "性能",
    "夜景算法",
    "机身"
  ],
  "summary": "该旗舰手机屏幕色彩准确、性能稳定，但存在夜景算法偏慢和机身发热明显的问题。"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

responses = model.batch(
    [
        "请解释 invoke()。",
        "请解释 stream()。",
        "请解释 batch()。",
    ],
    config={
        "run_name": "批量解释模型接口",
        "tags": ["models", "batch"],
        "metadata": {"user_id": "user-123"},
        "max_concurrency": 2,
    },
)