二次开发，在使用LangChain中的Tongyi模型进行流式输出streaming报错问题，官网框架的BUG修复。非常好的干货。

在使用LangChain中的Tongyi模型进行流式输出时，按照官方的代码直接运行会报一个类型错误：

TypeError: Additional kwargs key output_tokens already exists in left dict and value has unsupported type <class 'int'>.

​ 其指向的错误文件路径如下

C:\Users\Chenhao\AppData\Local\Programs\Python\Python39\lib\site-packages\langchain_core\utils\_merge.py

点开文件里面内容如下,其主要作用合并两个字典，然后处理字典中键值冲突的问题，之后将该BUG在stakflow上提问，得到了回复，根据测试直接让merge_dicts返回该字典即{'input_tokens': 530, 'output_tokens': 2, 'total_tokens': 532}，该BUG目前还没有修复。

from __future__ import annotations

from typing import Any, Dict


def merge_dicts(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]:
    """Merge two dicts, handling specific scenarios where a key exists in both
    dictionaries but has a value of None in 'left'. In such cases, the method uses the
    value from 'right' for that key in the merged dictionary.

    Example:
        If left = {"function_call": {"arguments": None}} and
        right = {"function_call": {"arguments": "{\n"}}
        then, after merging, for the key "function_call",
        the value from 'right' is used,
        resulting in merged = {"function_call": {"arguments": "{\n"}}.
    """
    merged = left.copy()
    for k, v in right.items():
        if k not in merged:
            merged[k] = v
        elif merged[k] is None and v:
            merged[k] = v
        elif v is None:
            continue
        elif merged[k] == v:
            continue
        elif type(merged[k]) != type(v):
            raise TypeError(
                f'additional_kwargs["{k}"] already exists in this message,'
                " but with a different type."
            )
        elif isinstance(merged[k], str):
            merged[k] += v
        elif isinstance(merged[k], dict):
            merged[k] = merge_dicts(merged[k], v)
        elif isinstance(merged[k], list):
            merged[k] = merged[k] + v
        else:
            raise TypeError(
                f"Additional kwargs key {k} already exists in left dict and value has "
                f"unsupported type {type(merged[k])}."
            )

    return merged

修改之后的文件如下,由于修改的是库文件中的代码，所以会有确认提示，在修改了之后在运行上流式输出的代码即可，前提是已经申请成果，且配置好了相关的API-KEY

from __future__ import annotations

from typing import Any, Dict

def merge_dicts(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]:
    merged = {'input_tokens': 530, 'output_tokens': 2, 'total_tokens': 532}
    return merged

修复之后

再次运行流式输出，完成。

from langchain_community.llms.tongyi import Tongyi

# 通义获取api key
from dotenv import find_dotenv, load_dotenv
import os
load_dotenv(find_dotenv())
DASHSCOPE_API_KEY = os.environ["DASHSCOPE_API_KEY"]

model = Tongyi(model_name='qwen-max', streaming=True)

for chunk in model.stream("讲一个翠花的故事。"):
    print(chunk, end="", flush=True)