LangChain(二)——Model I/O

LangChain模块架构图 [1]

一、Model I/O 简介

Model I/O三元组

  • PromptTemple：模板化、动态选择和管理模型输入
  • Language Models：通过通用接口调用LLM
  • OutputParser：从LLM输出中提取信息

数据流：Prompt->Model->Output Parser

二、模型API

2.1 OpenAI模型封装

1
2
3
4
5
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo")  # 默认是gpt-3.5-turbo
response = llm.invoke("你是谁")
print(response.content)

输出

1
我是一个人工智能助手，可以回答你的问题和提供帮助。你有什么想知道的或者需要帮忙的吗？

2.2 多轮对话Session封装

2.2.1. 消息组成

  • 角色（Role）：描述消息的发送者是谁。
  • 内容（Content）：消息的具体内容，可以是：
    • 字符串（大多数模型处理这种类型的内容）
    • 列表（包含字典），用于多模态输入，字典中包含输入类型和位置的信息。
  • 附加参数（additional_kwargs）：用于传递关于消息的额外信息，通常用于特定提供商的输入参数，而非通用参数。一个著名的例子是OpenAI的function_call。

2.2.2 消息类型

  • HumanMessage：代表用户的消息，通常只包含内容。
  • AIMessage：代表模型的消息，可能包含additional_kwargs，例如使用OpenAI工具调用时的tool_calls。
  • SystemMessage：代表系统消息，指示模型如何行为，通常只包含内容。并非所有模型都支持这种类型。
  • FunctionMessage：代表函数调用的结果。除了role和content，此消息还有一个name参数，表示产生此结果的函数名称。
  • ToolMessage：代表工具调用的结果。与FunctionMessage不同，为了匹配OpenAI的function和tool消息类型。除了role和content，此消息还有一个tool_call_id参数，表示产生此结果的工具调用的ID。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain.schema import (
    AIMessage,  # 等价于OpenAI接口中的assistant role
    HumanMessage,  # 等价于OpenAI接口中的user role
    SystemMessage  # 等价于OpenAI接口中的system role
)

messages = [
    SystemMessage(content="你是一个课程助理。"),
    HumanMessage(content="我叫小明。"),
    AIMessage(content="欢迎！"),
    HumanMessage(content="我是谁")
]

ret = llm.invoke(messages)

print(ret)

输出

1
content='您是小明先生。' response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 53, 'total_tokens': 60}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None} id='run-b3246018-5513-4b19-8959-08a70b6f2cf8-0'

2.3 通义千问

1
2
3
4
5
6
7
8
9
from langchain_community.chat_models import ChatTongyi
llm = ChatTongyi(model_name='qwen-max')
messages = [
    HumanMessage(content="你是谁")
]

ret = llm.invoke(messages)

print(ret.content)

输出

1
我是阿里云开发的一款超大规模语言模型，我叫通义千问。作为一个AI助手，我的设计目标是帮助用户获得准确、有用的信息，解答各种问题，提供知识性支持，进行富有成效的对话交流。在与您的互动中，我会运用我所学习的广泛知识和语言理解能力，竭诚为您提供服务。有什么我可以帮助您的吗？

2.4 Tool calling

支持工具调用特性的 LangChain ChatModels 实现了一个 .bind_tools 方法，该方法接收一个 LangChain 工具对象、Pydantic 类或 JSON Schemas 的列表，并以提供商特定的预期格式将它们绑定到聊天模型。

2.4.1 工具实现

LangChain Tool

使用 @tool 装饰器在 Python 函数上定义自定义工具的模式：

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
from langchain_core.tools import tool


@tool
def add(a: int, b: int) -> int:
    """Adds a and b.

    Args:
        a: first int
        b: second int
    """
    return a + b


@tool
def multiply(a: int, b: int) -> int:
    """Multiplies a and b.

    Args:
        a: first int
        b: second int
    """
    return a * b


tools = [add, multiply]

Pydantic class

也可以等效地使用 Pydantic 来定义模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from langchain_core.pydantic_v1 import BaseModel, Field


# Note that the docstrings here are crucial, as they will be passed along
# to the model along with the class name.
class add(BaseModel):
    """Add two integers together."""

    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")


class multiply(BaseModel):
    """Multiply two integers together."""

    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")


tools = [add, multiply]

2.4.2 Tool calling

使用 bind_tools() 方法来处理将 Multiply 转换为“工具”，并将其绑定到模型（即，每次调用模型时都传递它）。

工具调用包含在 LLM 响应中，它们将作为 .tool_calls 属性中的 ToolCall 对象列表附加到相应的 AIMessage 或 AIMessageChunk（流式传输时）。 ToolCall 是一个类型化字典，其中包括工具名称、参数值字典和（可选）标识符。 没有工具调用的消息默认为此属性的空列表。

1
2
3
4
5
6
7
8
9
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

llm_with_tools = llm.bind_tools(tools)

query = "What is 3 * 12? Also, what is 11 + 49?"

llm_with_tools.invoke(query).tool_calls

输出

1
2
3
4
5
6
[{'name': 'multiply',
  'args': {'a': 3, 'b': 12},
  'id': 'call_ZrGba27mD7sblS4b9wQLAomE'},
 {'name': 'add',
  'args': {'a': 11, 'b': 49},
  'id': 'call_SCJTaNMIgbV9REQWV5GqcHzH'}]

输出解析器可以进一步处理输出

1
2
3
4
from langchain_core.output_parsers.openai_tools import PydanticToolsParser

chain = llm_with_tools | PydanticToolsParser(tools=[multiply, add])
chain.invoke(query)

输出

1
[multiply(a=3, b=12), add(a=11, b=49)]

当仅使用bind_tools(tools)时，模型可以选择是否返回一个工具调用、多个工具调用或根本不返回工具调用。 某些模型支持 tool_choice 参数，该参数能够强制模型调用工具。 对于支持此功能的模型，可以传入希望模型始终调用 tool_choice="xyz_tool_name" 的工具名称。 或者您可以传入 tool_choice="any" 来强制模型调用至少一个工具，而无需具体指定哪个工具。

1
2
3
4
# 模型始终调用乘法工具
always_multiply_llm = llm.bind_tools([multiply], tool_choice="multiply")
# 始终至少调用加法或乘法之一
always_call_tool_llm = llm.bind_tools([add, multiply], tool_choice="any")

2.5 流式响应

当在流式上下文中调用工具时，消息块将通过 .tool_call_chunks 属性使用列表中的工具调用块对象填充。 ToolCallChunk 包含工具名称、参数和 id 的可选字符串字段，并包含可用于将块连接在一起的可选整数字段索引。

由于消息块继承自其父消息类，因此具有工具调用块的 AIMessageChunk 还将包含 .tool_calls 和 .invalid_tool_calls 字段。 这些字段是从消息的工具调用块中解析的。

1
2
async for chunk in llm_with_tools.astream(query):
    print(chunk.tool_call_chunks)

输出

1
2
3
4
5
6
7
8
9
10
11
12
[]
[{'name': 'multiply', 'args': '', 'id': 'call_5Gdgx3R2z97qIycWKixgD2OU', 'index': 0}]
[{'name': None, 'args': '{"a"', 'id': None, 'index': 0}]
[{'name': None, 'args': ': 3, ', 'id': None, 'index': 0}]
[{'name': None, 'args': '"b": 1', 'id': None, 'index': 0}]
[{'name': None, 'args': '2}', 'id': None, 'index': 0}]
[{'name': 'add', 'args': '', 'id': 'call_DpeKaF8pUCmLP0tkinhdmBgD', 'index': 1}]
[{'name': None, 'args': '{"a"', 'id': None, 'index': 1}]
[{'name': None, 'args': ': 11,', 'id': None, 'index': 1}]
[{'name': None, 'args': ' "b": ', 'id': None, 'index': 1}]
[{'name': None, 'args': '49}', 'id': None, 'index': 1}]
[]

添加消息块将合并其相应的工具调用块

1
2
3
4
5
6
7
8
9
10
11
12
first = True
async for chunk in llm_with_tools.astream(query):
    if first:
        gathered = chunk
        first = False
    else:
        gathered = gathered + chunk

    print(gathered.tool_call_chunks)


print(type(gathered.tool_call_chunks[0]["args"]))

输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
[]
[{'name': 'multiply', 'args': '', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}]
[{'name': 'multiply', 'args': '{"a"', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}]
[{'name': 'multiply', 'args': '{"a": 3, ', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 1', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 12}', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 12}', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}, {'name': 'add', 'args': '', 'id': 'call_GERgANDUbRqdtmXRbIAS9JTS', 'index': 1}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 12}', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}, {'name': 'add', 'args': '{"a"', 'id': 'call_GERgANDUbRqdtmXRbIAS9JTS', 'index': 1}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 12}', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}, {'name': 'add', 'args': '{"a": 11,', 'id': 'call_GERgANDUbRqdtmXRbIAS9JTS', 'index': 1}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 12}', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}, {'name': 'add', 'args': '{"a": 11, "b": ', 'id': 'call_GERgANDUbRqdtmXRbIAS9JTS', 'index': 1}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 12}', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}, {'name': 'add', 'args': '{"a": 11, "b": 49}', 'id': 'call_GERgANDUbRqdtmXRbIAS9JTS', 'index': 1}]
[{'name': 'multiply', 'args': '{"a": 3, "b": 12}', 'id': 'call_hXqj6HxzACkpiPG4hFFuIKuP', 'index': 0}, {'name': 'add', 'args': '{"a": 11, "b": 49}', 'id': 'call_GERgANDUbRqdtmXRbIAS9JTS', 'index': 1}]

<class 'str'>

部分解析

1
2
3
4
5
6
7
8
9
10
11
first = True
async for chunk in llm_with_tools.astream(query):
    if first:
        gathered = chunk
        first = False
    else:
        gathered = gathered + chunk

    print(gathered.tool_calls)

print(type(gathered.tool_calls[0]["args"]))

输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
[]
[]
[{'name': 'multiply', 'args': {}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}]
[{'name': 'multiply', 'args': {'a': 3}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 1}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 12}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 12}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 12}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}, {'name': 'add', 'args': {}, 'id': 'call_P39VunIrq9MQOxHgF30VByuB'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 12}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}, {'name': 'add', 'args': {'a': 11}, 'id': 'call_P39VunIrq9MQOxHgF30VByuB'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 12}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}, {'name': 'add', 'args': {'a': 11}, 'id': 'call_P39VunIrq9MQOxHgF30VByuB'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 12}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}, {'name': 'add', 'args': {'a': 11, 'b': 49}, 'id': 'call_P39VunIrq9MQOxHgF30VByuB'}]
[{'name': 'multiply', 'args': {'a': 3, 'b': 12}, 'id': 'call_aXQdLhKJpEpUxTNPXIS4l7Mv'}, {'name': 'add', 'args': {'a': 11, 'b': 49}, 'id': 'call_P39VunIrq9MQOxHgF30VByuB'}]


<class 'dict'>

三、Prompt 模板

3.1 Prompt 模板封装

  1. PromptTemplate 可以在模板中自定义变量
1
2
3
4
5
6
7
from langchain.prompts import PromptTemplate

template = PromptTemplate.from_template("给我写一首关于{subject}的诗")
print("===Template===")
print(template)
print("===Prompt===")
print(template.format(subject='黄河'))

输出

1
2
3
4
===Template===
input_variables=['subject'] template='给我写一首关于{subject}的诗'
===Prompt===
给我写一首关于黄河的诗
  1. ChatPromptTemplate 用模板表示的对话上下文
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
from langchain.prompts import (
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
    SystemMessagePromptTemplate,
)
from langchain_community.chat_models import ChatTongyi

llm = ChatTongyi(model_name='qwen-max')

template = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template(
            "你是{product}的客服助手。你的名字叫{name}"),
        HumanMessagePromptTemplate.from_template("{query}"),
    ]
)

prompt = template.format_messages(
    product="服装店",
    name="西瓜",
    query="你是谁"
)

ret = llm.invoke(prompt)

print(ret.content)

输出

1
我是西瓜，您的智能客服助手。我在这家时尚服装店中为您提供服务，帮助您解答关于商品、尺码、搭配建议、购物政策等方面的疑问，协助您挑选合适的服饰，确保您在购物过程中获得满意的支持与体验。有什么我可以帮您的吗？
  1. MessagesPlaceholder 把多轮对话变成模板
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
from langchain.prompts import (
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
    MessagesPlaceholder,
)
from langchain_core.messages import AIMessage, HumanMessage

human_prompt = "Translate your answer to {language}."
human_message_template = HumanMessagePromptTemplate.from_template(human_prompt)

chat_prompt = ChatPromptTemplate.from_messages(
    # variable_name 是 message placeholder 在模板中的变量名
    # 用于在赋值时使用
    [MessagesPlaceholder(variable_name="conversation"), human_message_template]
)

human_message = HumanMessage(content="Who is Elon Musk?")
ai_message = AIMessage(
    content="Elon Musk is a billionaire entrepreneur, inventor, and industrial designer"
)

messages = chat_prompt.format_prompt(
    # 对 "conversation" 和 "language" 赋值
    conversation=[human_message, ai_message], language="中文"
)

print(messages.to_messages())

result = llm.invoke(messages)
print(result.content)

输出

1
2
[HumanMessage(content='Who is Elon Musk?'), AIMessage(content='Elon Musk is a billionaire entrepreneur, inventor, and industrial designer'), HumanMessage(content='Translate your answer to 中文.')]
埃隆·马斯克是一位亿万富翁企业家、发明家和工业设计师

3.2 从文件加载 Prompt 模板

1
2
3
4
5
6
7
from langchain.prompts import PromptTemplate

template = PromptTemplate.from_file("example_prompt_template.txt")
print("===Template===")
print(template)
print("===Prompt===")
print(template.format(topic='三体'))

输出

1
2
3
4
===Template===
input_variables=['topic'] template='举一个关于{topic}的例子'
===Prompt===
举一个关于三体的例子

四、输出封装 OutputParser

自动把 LLM 输出的字符串按指定格式加载。

LangChain 内置的 OutputParser 包括:

  • ListParser
  • DatetimeParser
  • EnumParser
  • JsonOutputParser
  • PydanticParser
  • XMLParser

等等 [2]

4.1 Pydantic (JSON) Parser

自动根据 Pydantic 类的定义，生成输出的格式说明

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
from langchain_core.pydantic_v1 import BaseModel, Field, validator
from typing import List, Dict
    
from langchain.prompts import PromptTemplate, ChatPromptTemplate, HumanMessagePromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langchain_community.chat_models import ChatTongyi


# 定义你的输出对象


class Date(BaseModel):
    year: int = Field(description="Year")
    month: int = Field(description="Month")
    day: int = Field(description="Day")
    era: str = Field(description="BC or AD")

    # ----- 可选机制 --------
    # 你可以添加自定义的校验机制
    @validator('month')
    def valid_month(cls, field):
        if field <= 0 or field > 12:
            raise ValueError("月份必须在1-12之间")
        return field

    @validator('day')
    def valid_day(cls, field):
        if field <= 0 or field > 31:
            raise ValueError("日期必须在1-31日之间")
        return field

    @validator('day', pre=True, always=True)
    def valid_date(cls, day, values):
        year = values.get('year')
        month = values.get('month')

        # 确保年份和月份都已经提供
        if year is None or month is None:
            return day  # 无法验证日期，因为没有年份和月份

        # 检查日期是否有效
        if month == 2:
            if cls.is_leap_year(year) and day > 29:
                raise ValueError("闰年2月最多有29天")
            elif not cls.is_leap_year(year) and day > 28:
                raise ValueError("非闰年2月最多有28天")
        elif month in [4, 6, 9, 11] and day > 30:
            raise ValueError(f"{month}月最多有30天")

        return day

    @staticmethod
    def is_leap_year(year):
        if year % 400 == 0 or (year % 4 == 0 and year % 100 != 0):
            return True
        return False


model = ChatTongyi(model_name='qwen-max',top_p=0.1)

# 根据Pydantic对象的定义，构造一个OutputParser
parser = PydanticOutputParser(pydantic_object=Date)

template = """提取用户输入中的日期。
{format_instructions}
用户输入:
{query}"""

prompt = PromptTemplate(
    template=template,
    input_variables=["query"],
    # 直接从OutputParser中获取输出描述，并对模板的变量预先赋值
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

print("====Format Instruction=====")
print(parser.get_format_instructions())


query = "2077年八月6日，三体舰队打过来了"
model_input = prompt.format_prompt(query=query)

print("====Prompt=====")
print(model_input.to_string())

output = model.invoke(model_input.to_messages())
print("====模型原始输出=====")
print(output.content)
print("====Parse后的输出=====")
date = parser.parse(output.content)
print(date)
1
2
3
4
5
6
7
====Format Instruction=====
The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:

{"properties": {"year": {"title": "Year", "description": "Year","type": "integer"}, "month": {"title": "Month", "description": "Month","type": "integer"}, "day": {"title": "Day", "description": "Day","type": "integer"}, "era": {"title": "Era", "description": "BC or AD","type": "string"}}, "required": ["year", "month", "day", "era"]}

1
2
3
4
5
6
7
8
====Prompt=====
提取用户输入中的日期。
The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:

{"properties": {"year": {"title": "Year", "description": "Year", "type":"integer"}, "month": {"title": "Month", "description": "Month", "type":"integer"}, "day": {"title": "Day", "description": "Day", "type":"integer"}, "era": {"title": "Era", "description": "BC or AD", "type":"string"}}, "required": ["year", "month", "day", "era"]}
1
2
3
4
5
6
用户输入:
2077年八月6日，三体舰队打过来了
====模型原始输出=====
{"year": 2077, "month": 8, "day": 6, "era": "AD"}
====Parse后的输出=====
year=2077 month=8 day=6 era='AD'

4.2 Auto-Fixing Parser

利用 LLM 自动根据解析异常修复并重新解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain.output_parsers import OutputFixingParser

new_parser = OutputFixingParser.from_llm(
    parser=parser, llm= ChatTongyi(model_name='qwen-max'))

# 我们把之前output的格式改错
output = output.content.replace("8", "八月")
print("===格式错误的Output===")
print(output)
try:
    date = parser.parse(output)
except Exception as e:
    print("===出现异常===")
    print(e)

# 用OutputFixingParser自动修复并解析
date = new_parser.parse(output)
print("===重新解析结果===")
print(date.json())

输出

1
2
3
4
5
6
===格式错误的Output===
{"year": 2077, "month": 八月, "day": 6, "era": "AD"}
===出现异常===
Invalid json output: {"year": 2077, "month": 八月, "day": 6, "era": "AD"}
===重新解析结果===
{"year": 2077, "month": 8, "day": 6, "era": "AD"}

参考

LangChain
#LLM #LLM学习笔记 #LangChain
LangChain(二)——Model I/O
https://mztchaoqun.com.cn/posts/D26_LangChain_ModelIO/
作者
mztchaoqun
发布于
2024年6月28日
许可协议