learnNote

主题

LangChain 工具流式调用深度分析 ​

一、项目总体架构 ​

LangChain 是一个 Python monorepo，采用三层架构设计：

libs/
├── core/           # langchain-core (v1.2.16) —— 核心抽象层
├── langchain_v1/   # langchain (v1.2.10) —— 主包（Agent、中间件等）
├── langchain/      # langchain-classic（旧版，不再活跃开发）
└── partners/       # 15 个 Provider 集成（OpenAI、Anthropic、Ollama 等）

工具流式调用涉及的核心模块：

层级关键模块职责
工具定义层core/tools/BaseTool, StructuredTool, @tool 装饰器
消息类型层core/messages/ToolCall, ToolCallChunk, AIMessageChunk
模型层core/language_models/chat_models.pyBaseChatModel.stream/astream, bind_tools
Runnable 编排层core/runnables/base.pystream, astream, astream_events
回调/追踪层core/callbacks/, core/tracers/事件分发、流式事件生成
Agent 层langchain_v1/agents/, tools/tool_node.pyAgent 循环、工具执行中间件
Provider 集成层partners/openai/, partners/anthropic/具体 SSE 流的解析和 ToolCallChunk 生成

二、核心数据结构 ​

2.1 ToolCall —— 完整的工具调用请求 ​

python
# langchain_core/messages/tool.py
class ToolCall(TypedDict):
    name: str              # 工具名称
    args: dict[str, Any]   # 参数字典（已解析的 JSON）
    id: str | None         # 工具调用 ID（用于关联请求和响应）
    type: NotRequired[Literal["tool_call"]]

ToolCall 是 完整解析后 的工具调用，args 已经是 Python 字典。

2.2 ToolCallChunk —— 流式工具调用片段 ​

python
# langchain_core/messages/tool.py
class ToolCallChunk(TypedDict):
    name: str | None      # 工具名（可能在第一个 chunk 完整给出，后续为 None）
    args: str | None      # JSON 参数的字符串片段（如 '{"ci' 然后 'ty": "Paris"}'）
    id: str | None        # 工具调用 ID（通常在第一个 chunk 中给出）
    index: int | None     # ★ 合并索引：同一工具调用的所有 chunk 共享同一 index
    type: NotRequired[Literal["tool_call_chunk"]]

ToolCallChunk 是流式传输中的 原始字符串片段。index 字段是合并的关键标识。

2.3 ToolMessage —— 工具执行结果 ​

python
# langchain_core/messages/tool.py
class ToolMessage(BaseMessage, ToolOutputMixin):
    tool_call_id: str                            # 关联的工具调用 ID
    type: Literal["tool"] = "tool"
    artifact: Any = None                         # 非模型可见的额外产出
    status: Literal["success", "error"] = "success"

2.4 AIMessageChunk —— 流式 AI 消息 ​

python
# langchain_core/messages/ai.py
class AIMessageChunk(AIMessage, BaseMessageChunk):
    type: Literal["AIMessageChunk"] = "AIMessageChunk"
    
    tool_call_chunks: list[ToolCallChunk] = []   # 原始流式片段
    # 继承自 AIMessage:
    tool_calls: list[ToolCall] = []              # 自动尝试解析的完整调用
    invalid_tool_calls: list[InvalidToolCall] = []
    
    chunk_position: Literal["last"] | None = None  # "last" 表示流结束

核心设计：每个 AIMessageChunk 同时持有 tool_call_chunks（原始片段）和 tool_calls（用 parse_partial_json 自动尝试解析的结果）。

三、工具定义与注册 ​

3.1 BaseTool 基类 ​

python
# langchain_core/tools/base.py
class BaseTool(RunnableSerializable[str | dict | ToolCall, Any]):
    name: str                    # 工具名称
    description: str             # 工具描述（告诉 LLM 何时使用）
    args_schema: ArgsSchema | None  # Pydantic 模型或 JSON Schema dict
    response_format: Literal["content", "content_and_artifact"] = "content"

BaseTool 继承自 RunnableSerializable，这意味着它本身就是一个 Runnable，支持 invoke/ainvoke/stream/astream 等方法。

3.2 @tool 装饰器 ​

python
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气。"""
    return f"{city}: 晴天 25°C"

# 带参数的装饰器
@tool(response_format="content_and_artifact", parse_docstring=True)
def search(query: str) -> tuple[str, dict]:
    """搜索工具。
    Args:
        query: 搜索关键词。
    """
    return "结果摘要", {"full": "完整结果对象"}

@tool 装饰器内部流程：

  1. 通过 create_schema_from_function() 从函数签名自动推断 args_schema（Pydantic 模型）
  2. 创建 StructuredTool.from_function() 实例
  3. StructuredTool._run() 直接调用原始函数

3.3 bind_tools —— 将工具绑定到模型 ​

python
# langchain_core/language_models/chat_models.py (抽象接口)
class BaseChatModel:
    def bind_tools(
        self,
        tools: Sequence[dict | type | Callable | BaseTool],
        *,
        tool_choice: str | None = None,
        **kwargs,
    ) -> Runnable[LanguageModelInput, AIMessage]:
        raise NotImplementedError

具体实现由各 Provider 提供。例如 OpenAI：

python
# partners/openai/langchain_openai/chat_models/base.py
def bind_tools(self, tools, *, tool_choice=None, strict=None, 
               parallel_tool_calls=None, **kwargs):
    # 1. 转换为 OpenAI 工具格式
    formatted_tools = [convert_to_openai_tool(t, strict=strict) for t in tools]
    
    # 2. 处理 tool_choice（"auto"/"none"/"required"/工具名/dict）
    if isinstance(tool_choice, str) and tool_choice in ("any", True):
        tool_choice = "required"
    
    # 3. 绑定到模型参数
    return super().bind(tools=formatted_tools, tool_choice=tool_choice, **kwargs)

四、流式工具调用的完整链路 ​

4.1 全局数据流概览 ​

用户调用 model.stream(messages)
        │
        ▼
  BaseChatModel.stream()
    │  调用子类 _stream() 方法
    ▼
  Provider._stream()    <── 发起 SSE 流式请求
    │  逐事件解析 Provider 原始响应
    │  构造 AIMessageChunk(tool_call_chunks=[...])
    ▼
  yield AIMessageChunk ──────────────────────────→ 消费者逐个接收
    │                                                │
    │  收集所有 chunks                                ▼
    │                                          消费者用 __add__ 累加:
    │                                          accumulated = chunk1 + chunk2 + ...
    ▼                                                │
  yield 空 chunk (chunk_position="last")              ▼
    │                                          add_ai_message_chunks()
    ▼                                            ├─ merge_content(): 拼接文本
  merge_chat_generation_chunks(all_chunks)       ├─ merge_lists(tool_call_chunks): 按 index 合并
  → on_llm_end 回调                              │   └─ merge_dicts(): 字符串拼接 name/args
                                                 └─ init_tool_calls(): 自动触发
                                                     parse_partial_json(args)
                                                     → self.tool_calls（部分/完整解析）
                                                     
                                                 当 chunk_position="last" 时:
                                                   最终 parse → 完整的 tool_calls
                                                   content 中 tool_call_chunk → tool_call

4.2 BaseChatModel.stream() 的实现 ​

python
# langchain_core/language_models/chat_models.py
def stream(self, input, config=None, *, stop=None, **kwargs):
    if not self._should_stream(async_api=False, **{**kwargs, "stream": True}):
        # 模型不支持流式或 disable_streaming="tool_calling" 且有 tools
        yield cast("AIMessageChunk", self.invoke(input, config=config, stop=stop, **kwargs))
    else:
        # 设置回调管理器...
        chunks: list[ChatGenerationChunk] = []
        for chunk in self._stream(input_messages, stop=stop, **kwargs):
            if chunk.message.id is None:
                chunk.message.id = run_id
            chunk.message.response_metadata = _gen_info_and_msg_metadata(chunk)
            
            # 通过回调通知新 token
            run_manager.on_llm_new_token(
                cast("str", chunk.message.content), chunk=chunk
            )
            chunks.append(chunk)
            yield cast("AIMessageChunk", chunk.message)  # ← 逐个 yield

        # ★ 流结束后，补发一个 chunk_position="last" 的空 chunk
        if yielded and not chunk.message.chunk_position:
            msg_chunk = AIMessageChunk(
                content=empty_content, chunk_position="last", id=run_id
            )
            yield msg_chunk

        # 合并所有 chunks 用于 on_llm_end 回调（LangSmith 追踪等）
        generation = merge_chat_generation_chunks(chunks)
        run_manager.on_llm_end(LLMResult(generations=[[generation]]))

关键设计点：

  1. _should_stream() 判断：支持 disable_streaming="tool_calling" 选项。当设置此值且请求中包含 tools 参数时，自动回退到非流式 invoke，这在某些场景下可以简化工具调用处理。

  2. chunk_position="last" 终止信号：流结束时额外 yield 一个空 chunk 标记流终止，触发下游的最终工具调用解析。

  3. on_llm_new_token 回调：每个 chunk 都通过回调系统通知，驱动 astream_events 等高级流式 API。

五、ToolCallChunk 的流式合并机制 ​

5.1 AIMessageChunk.add() 合并入口 ​

python
# langchain_core/messages/ai.py
class AIMessageChunk(AIMessage, BaseMessageChunk):
    def __add__(self, other):
        if isinstance(other, AIMessageChunk):
            return add_ai_message_chunks(self, other)
        return super().__add__(other)

5.2 add_ai_message_chunks() 核心合并函数 ​

python
# langchain_core/messages/ai.py
def add_ai_message_chunks(left: AIMessageChunk, *others: AIMessageChunk):
    # 1. 合并 content（字符串拼接或 list merge）
    content = merge_content(left.content, *(o.content for o in others))

    # 2. 合并 additional_kwargs（递归 dict merge）
    additional_kwargs = merge_dicts(left.additional_kwargs, *(o.additional_kwargs ...))

    # 3. ★ 合并 tool_call_chunks —— 关键！
    if raw_tool_calls := merge_lists(
        left.tool_call_chunks, *(o.tool_call_chunks for o in others)
    ):
        tool_call_chunks = [
            create_tool_call_chunk(
                name=rtc.get("name"),
                args=rtc.get("args"),
                index=rtc.get("index"),
                id=rtc.get("id"),
            ) for rtc in raw_tool_calls
        ]
    else:
        tool_call_chunks = []

    # 4. 合并 usage_metadata（数值相加）
    usage_metadata = ...

    # 5. chunk_position: 任一 chunk 为 "last" 则结果也是 "last"
    chunk_position = "last" if any(
        x.chunk_position == "last" for x in [left, *others]
    ) else None

    return left.__class__(
        content=content,
        tool_call_chunks=tool_call_chunks,  # ← 合并后的 chunks
        chunk_position=chunk_position,
        ...
    )

5.3 merge_lists() —— 按 index 合并的核心算法 ​

python
# langchain_core/utils/_merge.py
def merge_lists(left: list | None, *others: list | None) -> list | None:
    merged = left.copy() if left is not None else None
    for other in others:
        if other is None:
            continue
        if merged is None:
            merged = other.copy()
        else:
            for e in other:
                if isinstance(e, dict) and "index" in e and isinstance(e["index"], int):
                    # 查找 merged 中 index 相同且 ID 不矛盾的元素
                    to_merge = [
                        i for i, e_left in enumerate(merged)
                        if "index" in e_left
                        and e_left["index"] == e["index"]
                        and (e_left.get("id") in (None, "")
                             or e.get("id") in (None, "")
                             or e_left.get("id") == e.get("id"))
                    ]
                    if to_merge:
                        # 找到了——用 merge_dicts 递归合并（字符串拼接！）
                        merged[to_merge[0]] = merge_dicts(merged[to_merge[0]], e)
                    else:
                        # 没找到——新的工具调用，追加
                        merged.append(e)
                else:
                    merged.append(e)
    return merged

合并规则：

  • 同一个 index 值的 chunk 被视为同一工具调用的片段，merge_dicts 会将字符串字段（name, args）拼接起来
  • 不同 index 值的 chunk 被视为不同的工具调用，直接追加到列表

流式合并示例：

Chunk 1: tool_call_chunks=[{name:"get_weather", args:'{"ci', id:"call_1", index:0}]
Chunk 2: tool_call_chunks=[{name:None, args:'ty": "Pa', id:None, index:0}]
Chunk 3: tool_call_chunks=[{name:None, args:'ris"}', id:None, index:0}]

accumulated = Chunk1 + Chunk2 + Chunk3
→ tool_call_chunks = [{name:"get_weather", args:'{"city": "Paris"}', id:"call_1", index:0}]
→ tool_calls = [{name:"get_weather", args:{"city":"Paris"}, id:"call_1"}]  # 自动解析

5.4 init_tool_calls() —— 自动解析验证器 ​

python
# langchain_core/messages/ai.py
class AIMessageChunk:
    @model_validator(mode="after")
    def init_tool_calls(self) -> Self:
        if not self.tool_call_chunks:
            return self
        
        # 每次创建 AIMessageChunk 时，自动尝试解析 tool_call_chunks
        tool_calls = []
        invalid_tool_calls = []
        for chunk in self.tool_call_chunks:
            try:
                # ★ 使用 parse_partial_json 容错解析
                args_ = parse_partial_json(chunk["args"]) if chunk["args"] else {}
                if isinstance(args_, dict):
                    tool_calls.append(
                        create_tool_call(name=chunk["name"] or "", args=args_, id=chunk["id"])
                    )
                else:
                    invalid_tool_calls.append(...)
            except Exception:
                invalid_tool_calls.append(...)
        
        self.tool_calls = tool_calls
        self.invalid_tool_calls = invalid_tool_calls
        
        # ★ 当 chunk_position="last" 时，将 content 中的 tool_call_chunk 替换为 tool_call
        if self.chunk_position == "last" and self.tool_call_chunks:
            id_to_tc = {tc["id"]: {...} for tc in self.tool_calls}
            for idx, block in enumerate(self.content):
                if block.get("type") == "tool_call_chunk" and block.get("id") in id_to_tc:
                    self.content[idx] = id_to_tc[call_id]
        
        return self

parse_partial_json 容错解析：即使 JSON 不完整（如 '{"city": "Pa'），也能尽可能解析出部分结果（如 {"city": "Pa"}）。这使得消费者在流式过程中就能看到逐步增长的 tool_calls。

六、Provider 集成层 —— SSE 事件到 ToolCallChunk ​

6.1 OpenAI —— Chat Completions API ​

python
# partners/openai/langchain_openai/chat_models/base.py

def _convert_delta_to_message_chunk(_dict, default_class):
    """将 OpenAI delta 转换为 AIMessageChunk"""
    tool_call_chunks = []
    if raw_tool_calls := _dict.get("tool_calls"):
        tool_call_chunks = [
            tool_call_chunk(
                name=rtc["function"].get("name"),     # 第一个 chunk 有值，后续为 None
                args=rtc["function"].get("arguments"), # JSON 字符串片段
                id=rtc.get("id"),                      # 第一个 chunk 有值
                index=rtc["index"],                    # ★ 合并标识
            )
            for rtc in raw_tool_calls
        ]
    
    return AIMessageChunk(
        content=content,
        tool_call_chunks=tool_call_chunks,
    )

OpenAI SSE 事件流示例：

json
// chunk 1: 工具调用开始
{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_abc123","function":{"name":"get_weather","arguments":""}}]}}]}

// chunk 2-N: 参数增量
{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"ci"}}]}}]}
{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ty\": \"Paris\"}"}}]}}]}

6.2 OpenAI —— Responses API ​

python
# 处理三种关键事件：

# 事件 1: response.output_item.added (function_call 类型)
# → 创建包含完整 name 和 call_id 的初始 chunk
tool_call_chunks.append({
    "type": "tool_call_chunk",
    "name": chunk.item.name,
    "args": chunk.item.arguments,
    "id": chunk.item.call_id,
    "index": current_index,
})

# 事件 2: response.function_call_arguments.delta
# → 增量参数片段，只有 args 和 index
tool_call_chunks.append({
    "type": "tool_call_chunk",
    "args": chunk.delta,
    "index": current_index,
})

# 事件 3: response.output_item.done (custom_tool_call 类型)
# → 完整的自定义工具调用
tool_call_chunks.append({
    "type": "tool_call_chunk",
    "name": chunk.item.name,
    "args": json.dumps({"__arg1": chunk.item.input}),
    "id": chunk.item.call_id,
    "index": current_index,
})

6.3 Anthropic ​

python
# partners/anthropic/langchain_anthropic/chat_models.py

def _make_message_chunk_from_anthropic_event(event, ...):
    # 事件 1: content_block_start + tool_use 类型
    # → 创建包含 id、name 的初始 chunk
    if event.type == "content_block_start" and "tool_use" in event.content_block.type:
        tool_call_chunk = create_tool_call_chunk(
            index=event.index,
            id=event.content_block.id,
            name=event.content_block.name,
            args="",  # 初始为空
        )
    
    # 事件 2: content_block_delta + input_json_delta 类型
    # → 增量 JSON 参数
    elif event.delta.type == "input_json_delta":
        tool_call_chunk = create_tool_call_chunk(
            index=event.index,
            id=None,      # 已在 start 事件中传递
            name=None,    # 已在 start 事件中传递
            args=event.delta.partial_json,
        )
    
    return AIMessageChunk(content=[content_block], tool_call_chunks=tool_call_chunks)

6.4 Provider 对比总结 ​

方面OpenAI (Chat Completions)OpenAI (Responses API)Anthropic
工具格式{"type":"function","function":{...}}原生 function tool{"name":..,"input_schema":{...}}
流式初始事件delta 中 tool_calls[].function.name + idresponse.output_item.addedcontent_block_start (type=tool_use)
参数增量delta 中 tool_calls[].function.argumentsresponse.function_call_arguments.deltacontent_block_delta (type=input_json_delta)
参数格式JSON 字符串片段chunk.delta 字符串partial_json 字符串
Chunk 构造tool_call_chunk(name, args, id, index)直接构造 dictcreate_tool_call_chunk(index, id, name, args)
最终载体AIMessageChunk.tool_call_chunksAIMessageChunk.tool_call_chunksAIMessageChunk.tool_call_chunks
上下文跟踪通过 index 字段通过 output_index + current_index通过 block_start_event 保存

统一出口：所有 Provider 最终都将流式工具调用信息标准化为 AIMessageChunk.tool_call_chunks 列表，下游消费者可以透明处理。

七、工具执行流程 ​

7.1 BaseTool.run() —— 同步执行 ​

python
# langchain_core/tools/base.py
class BaseTool(RunnableSerializable):
    def run(self, tool_input, *, config=None, tool_call_id=None, **kwargs):
        # 1. 配置回调管理器
        callback_manager = CallbackManager.configure(callbacks, self.callbacks, ...)
        
        # 2. 过滤注入参数，触发 on_tool_start 回调
        filtered_tool_input = self._filter_injected_args(tool_input)
        run_manager = callback_manager.on_tool_start(
            {"name": self.name, "description": self.description},
            tool_input_str, inputs=filtered_tool_input,
            tool_call_id=tool_call_id,
        )
        
        # 3. 解析输入参数（Pydantic 验证）
        tool_args, tool_kwargs = self._to_args_and_kwargs(tool_input, tool_call_id)
        
        # 4. 执行工具函数
        response = self._run(*tool_args, **tool_kwargs)
        
        # 5. 格式化输出
        if self.response_format == "content_and_artifact":
            content, artifact = response
        else:
            content = response
        
        # 6. 包装为 ToolMessage（如果有 tool_call_id）
        output = _format_output(content, artifact, tool_call_id, self.name, status)
        
        # 7. 触发 on_tool_end 回调
        run_manager.on_tool_end(output)
        return output

7.2 invoke/ainvoke 入口 ​

python
class BaseTool:
    def invoke(self, input: str | dict | ToolCall, config=None, **kwargs):
        tool_input, kwargs = _prep_run_args(input, config, **kwargs)
        return self.run(tool_input, **kwargs)

_prep_run_args 会处理 ToolCall 类型输入，提取 tool_call_id 和 args：

python
def _prep_run_args(value, config, **kwargs):
    if _is_tool_call(value):  # value.get("type") == "tool_call"
        tool_call_id = value["id"]
        tool_input = value["args"].copy()
    else:
        tool_call_id = None
        tool_input = value
    return tool_input, dict(callbacks=..., tool_call_id=tool_call_id, ...)

7.3 _format_output —— 输出格式化 ​

python
def _format_output(content, artifact, tool_call_id, name, status):
    if isinstance(content, ToolOutputMixin) or tool_call_id is None:
        return content  # 直接返回
    if not _is_message_content_type(content):
        content = json.dumps(content, ensure_ascii=False)  # JSON 序列化
    return ToolMessage(
        content, artifact=artifact,
        tool_call_id=tool_call_id, name=name, status=status,
    )

八、Runnable 流式编排 ​

8.1 stream/astream 基础 ​

python
# langchain_core/runnables/base.py
class Runnable:
    def stream(self, input, config=None, **kwargs) -> Iterator[Output]:
        """默认实现：直接 yield invoke 结果"""
        yield self.invoke(input, config, **kwargs)
    
    async def astream(self, input, config=None, **kwargs) -> AsyncIterator[Output]:
        """默认实现：直接 yield ainvoke 结果"""
        yield await self.ainvoke(input, config, **kwargs)

子类（如 BaseChatModel、RunnableSequence）覆盖此方法实现真正的流式输出。

8.2 RunnableSequence.stream() —— 链式流式 ​

python
class RunnableSequence(RunnableSerializable):
    def stream(self, input, config=None, **kwargs):
        yield from self.transform(iter([input]), config, **kwargs)
    
    def transform(self, input, config=None, **kwargs):
        # 将输入依次传递给链中的每个步骤
        # 中间步骤通过 transform 传递流式数据
        # 最后一个步骤的输出直接 yield

8.3 astream_events —— 事件流式 API ​

astream_events 是 LangChain 提供的高级流式 API，可以观察整个 Runnable 链执行过程中的所有中间事件：

python
class Runnable:
    async def astream_events(
        self, input, config=None, *,
        version="v2",
        include_names=None, include_types=None, include_tags=None,
        exclude_names=None, exclude_types=None, exclude_tags=None,
        **kwargs,
    ) -> AsyncIterator[StreamEvent]:
        event_stream = _astream_events_implementation_v2(
            self, input, config=config,
            include_names=..., exclude_names=..., **kwargs,
        )
        async with aclosing(event_stream):
            async for event in event_stream:
                yield event

StreamEvent 类型定义：

python
# langchain_core/runnables/schema.py
class EventData(TypedDict, total=False):
    input: Any              # Runnable 的输入
    output: Any             # Runnable 的输出
    chunk: Any              # 流式 chunk
    error: BaseException    # 执行异常
    tool_call_id: str | None  # 工具调用 ID（仅 on_tool_error）

class StandardStreamEvent(BaseStreamEvent):
    event: str         # 如 "on_tool_start", "on_chat_model_stream"
    name: str          # Runnable 名称
    run_id: str        # 执行 ID
    parent_ids: list[str]  # 父级运行 ID 链
    data: EventData    # 事件数据

工具相关的事件类型：

事件触发时机data 内容
on_tool_start工具开始执行{"input": {"x": 1, "y": 2}}
on_tool_end工具执行完成{"output": ToolMessage(...), "input": {...}}
on_tool_error工具执行出错{"error": Exception, "input": {...}, "tool_call_id": "..."}
on_chat_model_streamLLM 流式输出 chunk{"chunk": AIMessageChunk(...)}

九、回调系统与流式集成 ​

9.1 回调生命周期 ​

CallbackManager.on_tool_start()
    │  触发所有注册的 handler 的 on_tool_start
    │  返回 CallbackManagerForToolRun 实例
    ▼
[工具函数执行]
    │
    ├─ 成功 → run_manager.on_tool_end(output)
    │           触发所有 handler 的 on_tool_end
    │
    └─ 失败 → run_manager.on_tool_error(error)
                触发所有 handler 的 on_tool_error

9.2 _AstreamEventsCallbackHandler ​

这是 astream_events 的底层实现，继承 AsyncCallbackHandler：

python
# langchain_core/tracers/event_stream.py
class _AstreamEventsCallbackHandler(AsyncCallbackHandler, _StreamingCallbackHandler):
    def __init__(self, ...):
        self.run_map: dict[UUID, RunInfo] = {}
        memory_stream = _MemoryStream[StreamEvent](loop)
        self.send_stream = memory_stream.get_send_stream()
        self.receive_stream = memory_stream.get_receive_stream()
    
    async def on_tool_start(self, serialized, input_str, *, run_id, ...):
        """工具开始时发送 on_tool_start 事件"""
        self._write_run_start_info(run_id, ..., run_type="tool",
                                    inputs=inputs, tool_call_id=kwargs.get("tool_call_id"))
        self._send({
            "event": "on_tool_start",
            "data": {"input": inputs or {}},
            "name": name_, "run_id": str(run_id),
            "parent_ids": self._get_parent_ids(run_id),
        }, "tool")
    
    async def on_tool_end(self, output, *, run_id, **kwargs):
        """工具结束时发送 on_tool_end 事件"""
        run_info, inputs = self._get_tool_run_info_with_inputs(run_id)
        self._send({
            "event": "on_tool_end",
            "data": {"output": output, "input": inputs},
            ...
        }, "tool")
    
    async def on_tool_error(self, error, *, run_id, **kwargs):
        """工具出错时发送 on_tool_error 事件，包含 tool_call_id"""
        tool_call_id = kwargs.get("tool_call_id") or run_info.get("tool_call_id")
        self._send({
            "event": "on_tool_error",
            "data": {"error": error, "input": inputs, "tool_call_id": tool_call_id},
            ...
        }, "tool")
    
    async def tap_output_aiter(self, run_id, output):
        """拦截流式输出，生成 on_xxx_stream 事件"""
        async for chunk in output:
            self._send({
                "event": f"on_{run_info['run_type']}_stream",
                "data": {"chunk": chunk},
                ...
            }, run_info["run_type"])
            yield chunk

事件通过 _MemoryStream 异步管道传递：_send() 写入 send_stream，消费者通过 receive_stream 异步迭代读取。

9.3 StreamingStdOutCallbackHandler ​

最简单的流式回调，将 LLM token 直接打印到标准输出：

python
class StreamingStdOutCallbackHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs):
        sys.stdout.write(token)
        sys.stdout.flush()
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        pass  # 空实现
    
    def on_tool_end(self, output, **kwargs):
        pass  # 空实现

十、Agent 层的工具调用与流式 ​

10.1 ToolNode（LangGraph 集成） ​

python
# langchain_v1/langchain/tools/tool_node.py
from langgraph.prebuilt import InjectedState, InjectedStore, ToolRuntime
from langgraph.prebuilt.tool_node import ToolNode as _ToolNode

ToolNode 实际来自 LangGraph，是 Agent 图中执行工具调用的关键节点。它：

  • 从 AIMessage.tool_calls 中提取工具调用请求
  • 查找并执行对应的 BaseTool
  • 将结果包装为 ToolMessage 返回

10.2 Agent 工具调用中间件 ​

LangChain v1 提供了一套丰富的中间件来增强工具调用行为：

ToolCallLimitMiddleware —— 调用次数限制 ​

python
# langchain_v1/langchain/agents/middleware/tool_call_limit.py
class ExitBehavior(str, Enum):
    CONTINUE = "continue"  # 阻止被限制的工具但允许其他继续
    ERROR = "error"        # 抛出异常
    END = "end"            # 立即终止执行

class ToolCallLimitState(AgentState):
    thread_tool_call_count: dict[str, int]  # 线程级计数
    run_tool_call_count: dict[str, int]     # 运行级计数

ToolRetryMiddleware —— 工具重试 ​

python
class ToolRetryMiddleware(AgentMiddleware):
    max_attempts: int = 2       # 最大重试次数
    # 支持指数退避、自定义异常过滤、自定义错误消息格式

LLMToolEmulator —— LLM 模拟工具 ​

python
class LLMToolEmulator(AgentMiddleware):
    """使用 LLM 模拟工具执行（用于测试）"""
    model: str = "anthropic:claude-sonnet-4-5-20250929"
    tools: list[str] | None = None  # None 表示模拟所有工具

LLMToolSelectorMiddleware —— 工具选择 ​

python
class LLMToolSelectorMiddleware(AgentMiddleware):
    """使用独立 LLM 在调用主模型前筛选最相关的工具"""

十一、完整流式工具调用示例 ​

11.1 基础流式调用 ​

python
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气。"""
    return f"{city}: 晴天 25°C"

model = ChatOpenAI(model="gpt-4o").bind_tools([get_weather])

# 基础流式消费
full = None
for chunk in model.stream("北京天气怎么样？"):
    print(chunk)
    # 每个 chunk 是 AIMessageChunk
    # chunk.tool_call_chunks = [ToolCallChunk(...)]  # 原始片段
    # chunk.tool_calls = [ToolCall(...)]              # 自动尝试解析
    full = chunk if full is None else full + chunk

# full.tool_calls 包含完整解析的工具调用
print(full.tool_calls)
# [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_xxx'}]

11.2 使用 astream_events ​

python
async for event in model.astream_events("北京天气怎么样？", version="v2"):
    if event["event"] == "on_chat_model_stream":
        chunk = event["data"]["chunk"]
        if chunk.tool_call_chunks:
            print(f"工具调用片段: {chunk.tool_call_chunks}")
    elif event["event"] == "on_tool_start":
        print(f"工具开始: {event['name']}, 输入: {event['data']['input']}")
    elif event["event"] == "on_tool_end":
        print(f"工具结束: {event['name']}, 输出: {event['data']['output']}")

11.3 Agent + 流式工具调用（LangGraph） ​

python
from langchain.agents import create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

# 配合 LangGraph 使用
# ToolNode 从 AIMessage.tool_calls 提取调用 → 执行工具 → 返回 ToolMessage
# Agent 循环持续到 LLM 不再产生 tool_calls

十二、架构设计总结 ​

12.1 分层抽象 ​

┌──────────────────────────────────────────┐
│          消费者 / 应用层                    │
│  stream() / astream() / astream_events()  │
├──────────────────────────────────────────┤
│          Runnable 编排层                   │
│  RunnableSequence / RunnableParallel      │
├──────────────────────────────────────────┤
│          模型层                            │
│  BaseChatModel.stream() → _stream()       │
├──────────────────────────────────────────┤
│          Provider 集成层                   │
│  OpenAI._stream() / Anthropic._stream()   │
│  SSE 事件 → AIMessageChunk + ToolCallChunk│
├──────────────────────────────────────────┤
│          消息/工具数据层                    │
│  AIMessageChunk, ToolCallChunk, ToolCall  │
│  merge_lists(), init_tool_calls()         │
├──────────────────────────────────────────┤
│          回调/追踪层                       │
│  CallbackManager, _AstreamEventsHandler   │
│  on_tool_start / on_tool_end / on_tool_error│
└──────────────────────────────────────────┘

12.2 核心设计原则 ​

  1. 统一抽象：所有 Provider 的流式工具调用最终都标准化为 AIMessageChunk.tool_call_chunks，消费者无需关心具体 Provider

  2. 渐进解析：通过 parse_partial_json 和 init_tool_calls 验证器，每个 AIMessageChunk 在创建时就尝试解析 tool_calls，让消费者在流式过程中就能获取部分解析的结果

  3. 基于 index 的合并：ToolCallChunk.index 字段统一了不同 Provider 的合并策略，merge_lists 通过 index 匹配将属于同一工具调用的字符串片段拼接

  4. 终止信号：chunk_position="last" 标记流结束，触发最终的工具调用解析和 content 格式转换

  5. Runnable 协议：BaseTool 继承 RunnableSerializable，工具本身就是 Runnable，可以无缝参与 LCEL 链式组合

  6. 回调驱动的事件流：astream_events 通过回调系统（_AstreamEventsCallbackHandler）拦截所有内部事件，包括工具的 start/end/error，通过 _MemoryStream 异步管道传递给消费者

  7. 中间件模式：Agent 层通过可组合的中间件（限流、重试、模拟、选择）增强工具调用行为，不影响核心流式传输逻辑

最后更新: