该笔记是基于黑马的《大模型RAG与Agent智能体项目实战教程，基于主流的LangChain技术从大模型提示词到实战项目》课程编写的

一、OpenAI库基础使用

1.1  OpenAI库的基础使用

        0penAI库是0penAI官方推出的Python SDK，核心作用是让开发者能简单、高效地调用0penAI的各类API(如GPT聊天、DALL.E绘图、语音转文字等)，无需手动处理HTTP请求、身份验证等底层细节。
        由于其发布较早且比较易用，现如今许多模型服务商(如阿里云百炼平台)均兼容0penAISDK的调用。

1.1.1 获取客户端对象

from openai import OpenAI

client = OpenAI(
    api_key="your_api_key_here",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

        主要是用如上2个参数：（1）api_key：模型服务商提供的APIKEY密钥。（2）base_url：模型服务商的API接入地址，主要基于此参数来切换不同的模型服务商（如OpenAI、阿里云、腾讯云等）。

1.1.2 调用模型

from openai.types.chat.chat_completion import ChatCompletion
response:ChatCompleton=client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role":"system","content":"你是一个Python编程专家"},
        {"role":"assistant","content":"好的，我是编程专家，你要问什么？"},
        {"role":"user","content":"使用Python代码输出1-10的数字"}
    ]
)

        client.chat.completions.create创建ChatCompletion对象。

        主要参数有2个：①model:选择所用模型，如代码的qwen3-max；②messages：提供给模型的消息。（类型:list，可以包含多个字典消息，每个字典消息包含2个key，role:角色和content:内容。system角色:设定助手的整体行为、角色和规则，为对话提供上下文框架(如指定助手身份、回答风格、核心要求)，是全局的背景设定，影响后续所有交互。assistant角色:代表AI助手的回答，可以在代码中人为设定。user角色:代表用户，发送问题、指令或需求。

1.1.3 处理结果

        response变量就是ChatCompletion对象，其包含信息如下：

{
    "id": "chatcmpl-xxxx",
    "object": "chat.completion",
    "created": 1735689600,
    "model": "gpt-3.5-turbo-0125",
    "choices": [
    {
    "index": 0,
    "message": {
    "role": "assistant",
    "content": "生成的回复内容"
    },
    "finish_reason": "stop" # stop=正常结束，length=令牌数超限，function_call=触发函数调用
    }
    ],
    "usage": { # 令牌消耗统计
    "prompt_tokens": 50,
    "completion_tokens": 80,
    "total_tokens": 130
    }
}

        可以通过print(responce.choices[0].message.content)输出模型给出的回答信息。

from openai import OpenAI

#1.获取client对象，OpenAI类对象
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

#2.调用模型
response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role":"system","content":"你是一个Python编程专家，并且不说废话简单回答"},
        {"role":"assistant","content":"好的，我是编程专家，并且话不多，你要问什么？"},
        {"role":"user","content":"使用Python代码输出1-10的数字"}
    ]
)

#3.处理结果
print(response.choices[0].message.content)

1.2 OpenAI库的流式输出

        可以设定结果输出为stream模式（流式输出），获得更好的使用体验。开启流式输出主要就2步：（1）在client.chat.completions.create()调用模型的时候设定参数：stream=True；（2）for循环response对象，并在循环内输出内容。

from openai import OpenAI

#1.获取client对象，OpenAI类对象
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

#2.调用模型
response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role":"system","content":"你是一个Python编程专家，并且话非常多"},
        {"role":"assistant","content":"好的，我是编程专家，并且话非常多，你要问什么？"},
        {"role":"user","content":"使用Python代码输出1-10的数字"}
    ],
    stream=True  #开启流式输出功能
)

#3.处理结果
for chunk in response:
    print(
        chunk.choices[0].delta.content,
        end=" ",     #每一段之间以空格分隔
        flush=True   #立刻刷新缓冲区
 )

1.3 OpenAI库附带历史消息调用模型

        调用模型传入的参数messages，其要求是list对象，即表明其支持非常多的消息在内。基于此，将历史消息填入，让模型知晓对话的上下文，更好的回答。

from openai import OpenAI

#1.获取client对象，OpenAI类对象
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

#2.调用模型
response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role":"system","content":"你是AI助理，回答很简洁"},
        {"role":"user","content":"小明有2条宠物狗"},
        {"role":"assistant","content":"好的"},
        {"role":"user","content":"小红有3只宠物猫"},
        {"role":"assistant","content":"好的"},
        {"role":"user","content":"总共有几个宠物？"}
    ],
    stream=True  #开启流式输出功能
)

#3.处理结果
for chunk in response:
    print(
        chunk.choices[0].delta.content,
        end=" ",     #每一段之间以空格分隔
        flush=True   #立刻刷新缓冲区
 )

        当前的历史消息是一次性的，如果是生产系统可以将消息保存到文件、数据库等持久化工具内，需要的时候提取使用。

二、提示词工程

2.1  概念

        提示工程（Prompt Engineering），也称为In-Context Prompting，是指在不更新模型权重的情况下如何与大模型交互以引导其行为以获得所需结果的结果。

        在人工智能领域，Prompt指的是用户给大型语言模型发出的指令。例如，"「讲个笑话」"、 "「用Python编个贪吃蛇游戏」"、"「写封情书」"等。虽然看似简单，但实际上，Prompt的设计对于模型的结果影响很大。因此如何设计prompt，进而与模型更好的交互，是研究人员必备的必不可少的技能(提示工程)。

2.2 提示词技巧

        任何Prompt技巧，都不如清晰的表达你的需求。这就类似人与人沟通，如果话说不明白，不可能让别人理解你的思想。因此，写出清晰的指令，是核心。

        技巧1：详细的描述

        技巧2：让模型充当某个角色

        技巧3：使用分隔符标明输入的不同部分
        中括号、XML标签、三引号等分隔符可以帮助划分要区别对待的文本，也可以帮助模型更好的理解文本内容。常用“'''”把内容框起来。

        技巧4：对任务指定步骤

        对于可以拆分的任务可以尽量拆开，最好能为其指定一系列步骤，明确步骤可以让模型更容易实现它们。

        技巧5：提供例子

        本质类似于few-shot leaning。先扔给大模型举例，然后让模型按照例子来输出。

        技巧6：使用参考文本作答

        基于文本文档，辅助大模型问答，降低模型“幻觉”（一本正经的胡说八道）问题。即使用参考文本作答，经典的知识库用法，让大模型使用我们提供的信息来组成答案。

2.3 实战案例（金融文本分类）

2.3.1 背景

        当前金融领域信息化发展的时代,金融数据大量激增,许多投资者和研究者试图通过对这些数据进行深度分析而获得一些有效的决策和帮助，尽可能减少决策失误带来的损失。所以,针对金融数据的分析方法研究是目前十分有益且热门的话题。
        当前案例主要有三大业务场景实现:（1）基于大模型完成:金融文本分类；（2）基于大模型完成:金融文本信息抽取；（3）基于大模型完成:金融文本匹配。

        大模型选择:Qwen在线大模型(阿里云通义千问qwen3-max)。
        采用方法:基于Few-Shot+Zero-Shot的思想，设计prompt(提示词)，进而应用大模型完成相应的任务。

2.3.2 Zero-shot思想

        Zero-shot学习（Zero-shot Learning）是指在训练阶段不存在与测试阶段完全相同的类别，但是模型可以使用训练过的知识来推广到测试集中的新类别上。这种能力被成为“零样本”学习，因为模型在训练时从未见过测试集中的新类别，在模型训练和提示词优化中均有体现。

        在模型训练中：已知马(四脚兽)、虎(有条纹)、熊猫(黑白色)的特征，但未训练过斑马的数据(不认识)；告知模型:斑马是四脚兽、有黑白色的条纹；模型可以在已知数据中进行推理，从而识别斑马。

        在提示词优化中：Zero-shot思想用于基于已训练的能力，不提供任何示例，仅通过语言去描述任务的要求、目标和约束，让模型直接生成结果。简单来说就是“用语言定义任务，解放(信任)模型的预训练知识”。比如:请判断"“包围的用户评论中的情感倾向，输出正面或负面。"这款代餐鸡胸肉饱腹感很强，吃起来也不柴，很推荐!””

2.3.3 Few-shot思想

        Few-shot学习(Few-shot Learning)是指少样本学习，当模型在学习了一定类别的大量数据后，对于新的类别，只需要少量的样本就能快速学习，对应的有one-shot learning，单样本学习，也算样本少到为一的情况下的一种few-shot learning。

        在模型训练中(相似度判断方法)：基于少量企鹅样本并结合相识度判断，推论未知图片内含“企鹅”

        在提示词优化中：Few-shot主要用于基于少量示例，让模型参考示例回答。简单来说就是“用示例定义任务，在模型的预训练知识的基础上，提升模型回答的对齐精度(比如参考示例的格式)"。
        比如：请抽取产品名称和核心卖点2个字段，格式为Json，我提供2个示例。
        示例1:MacBookPro高效节能，性能强大，适合牛马工作使用                                                            输出:{“产品名称”:“MacBookPro”，“产品卖点”:“高效节能，性能强大”}                                                示例2:联想笔记本拥有RTX4060独立显卡，畅玩游戏，丝滑流畅                                                      输出:{“产品名称”:“联想笔记本，“产品卖点”:“畅玩游戏，丝滑流畅”}
        请处理:华为MatepadPro，高清大屏，长效续航，你的好帮手。
        {“产品名称”:“华为MatepadPro”，“产品卖点”:“高清大屏，长效续航”} 

2.3.4 Prompt设计

        对于大模型来讲，prompt 的设计非常重要，一个明确的prompt 能够帮助我们更好从大模型中获得我们想要的结果。

        在该任务的prompt设计中，我们主要考虑2点：一是需要向模型解释什么叫作「文本分类任务」；
二是需要让模型按照我们指定的格式输出。
        为了让模型知道什么叫做「文本分类」，我们借用FewShot的方式，给模型展示一些正确的例子:
User"今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。"是[新闻报道’'公司公告’'财务公告'分析师报告门里的什么类别?
Bot:新闻报道
User"本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。"是[新闻报道’'公司公告，'财务公告'分析师报告’]里的什么类别?
Bot:财务报告
        其中，User代表我们输入给模型的句子，Bot代表模型的回复内容。
        注意:上述例子中Bot的部分也是由人工输入的，其目的是希望看到在看到类似User中的句子时，模型应当做出类似Bot的问答。

from openai import OpenAI

# 1. 获取client对象，OpenAI类对象
client = OpenAI(
    # base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    base_url="http://localhost:11434/v1"
)

examples_data = {       # 示例数据
    '新闻报道': '今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。',
    '财务报告': '本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。',
    '公司公告': '本公司高兴地宣布成功完成最新一轮并购交易，收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域，提高市场竞争力',
    '分析师报告': '最新的行业分析报告指出，科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素，投资者应关注这些趋势'
}
# 分类列表
examples_types = ['新闻报道', '财务报道', '公司公告', '分析师报告']

# 提问数据
questions = [
    "今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。",
    "ABC公司今日发布公告称，已成功完成对XYZ公司股权的收购交易。本次交易是ABC公司在扩大业务范围、加强市场竞争力方面的重要举措。据悉，此次收购将进一步巩固ABC公司在行业中的地位，并为未来业务发展提供更广阔的发展空间。详情请见公司官方网站公告栏",
    "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。",
    "最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会",
    "小明喜欢小新哟"
]

"""
[
    {"role": "system",      "content": "你是金融专家，将文本分类为['新闻报道', '财务报道', '公司公告', '分析师报告']，不清楚的分类为'不清楚类别' 下面有示例："},
     
    {"role": "user",        "content": "今日，央行发布公告宣布降............."},
    {"role": "assistant",   "content": "新闻报道"},
    {"role": "user",        "content": "ABC公司今日发布公告称，已成功完成对XYZ公司股................."},
    {"role": "assistant",   "content": "财务报告},
    {"role": "user",        "content": "公司资产负债表显示，公司偿债能力强劲，现金流充足..................."},
    {"role": "assistant",   "content": "公司公告"},
    {"role": "user",        "content": "最新的分析报告指出，可再生能源............."},
    {"role": "assistant",   "content": "分析师报告"},
    
    {"role": "user",        "content": "要提问的问题"}
]
"""

messages = [
    {"role": "system", "content": "你是金融专家，将文本分类为['新闻报道', '财务报道', '公司公告', '分析师报告']，不清楚的分类为'不清楚类别' 下面有示例："},
]

for key, value in examples_data.items():
    messages.append({"role": "user", "content": value})
    messages.append({"role": "assistant", "content": key})


# 向模型提问
for q in questions:
    response = client.chat.completions.create(
        # model="qwen3-max",
        model="qwen3:4b",
        messages=messages + [{"role": "user", "content": f"按照示例，回答这段文本的分类类别：{q}"}]
    )

    print(response.choices[0].message.content)

2.4 JSON数据格式

2.4.1 基本概念

        JSON(JavaScript Object Notation)是一种轻量级的数据交换格式，易于人阅读和编写，同时也易于机器解析和生成。Json是带有格式的字符串，主要用于数据交换，即程序和程序之间的信息互传，使用Json会更加方便，如下示例:

        Json主要有2种结构：Json对象和Json数组。

2.4.2 Python中使用Json

        Python中使用Json主要完成：（1）将Python字典、列表转换为Json字符串；（2）读取Json字符串，转换为Python字典或列表。
        主要使用Python内置的json库：（1）json.dumps(字典或列表,ensure_ascii=False):将字典或列表转换为Json字符串（ensure_ascii参数确保中文能正常显示，返回值:Json字符串）；（2）json.loads(json字符串):将Json字符串转换为Python字典或列表（返回值:Python字典 或 Python列表）。

import json

d = {
    "name":"周杰伦",
    "age":11,
    "gender":"男"
}

s = json.dumps(d, ensure_ascii=False)
print(s)

l = [
    {
        "name":"周杰伦",
        "age":11,
        "gender":"男"
    },
    {
        "name":"章若楠",
        "age":18,
        "gender":"女"
    },
    {
        "name":"小明",
        "age":15,
        "gender":"男"
    }
]

print(json.dumps(l, ensure_ascii=False))

json_str = '{"name": "周杰伦", "age": 11, "gender": "男"}'
json_array_str = '[{"name": "周杰伦", "age": 11, "gender": "男"}, {"name": "章若楠", "age": 18, "gender": "女"}, {"name": "小明", "age": 15, "gender": "男"}] '

res_dict = json.loads(json_str)
print(res_dict, type(res_dict))

res_list = json.loads(json_array_str)
print(res_list, type(res_list))

2.5 项目实战（金融文本信息抽取）

2.5.1 任务介绍

        首先，我们定义信息抽取的Schema：

# 定义不同实体下的具备属性
schema = {
    '金融': ['日期', '股票名称', '开盘价', '收盘价', '成交量'],
}

下面几段文本来自某平台发布的股票信息：

1. '2023-02-15, 寓意吉祥的节日, 股票佰驾 [BD] 美股开盘价10美元, 虽然经历了波动, 但最终以13美元收盘, 成交量微幅增加至460,000, 投资者情绪较为平稳。',
2. '2023-04-05, 市场迎来轻松氛围, 股票盘古 (0021) 开盘价23元, 尽管经历了波动, 但最终以26美元收盘, 成交量缩小至310,000, 投资者保持观望态度。',

我们的目的是期望模型能够帮助我们识别出这2段话中的SPO三元组信息。

2.5.2 Prompt设计     

        在该任务的 prompt 设计中，我们主要考虑 2 点：（1）需要向模型解释什么叫作「信息抽取任务」；（2）需要让模型按照我们指定的格式（json）输出。

        为了让模型知道什么叫做「信息抽取」，我们借用 FewShot 的方式，先给模型展示几个正确的例子：

User:'2023-01-10, 股市震荡。股票占导-D[EOOE]美股今日开盘价100美元，一度飙升至105美元，随后回落至98美元，最终以102美元收盘，成交量达到520000。' 提取上述句子中"金融"（日期，‘股票名称’，‘开盘价’，‘收盘价’，‘成交量’）类型的实体，并按照JSON格式输出，上述句子中没有的信息用[原文中未提及]来表示，多个值之间用','分隔。

Bot:['日期':['2023-01-10'],'股票名称':['古哥-D[EOOE]美股'],'开盘价':['100美元'],'收盘价':['102美元'],'成交量':['520000']...

        其中，User 代表我们输入给模型的句子，Bot 代表模型的回复内容。

        注意：上述例子中 Bot 的部分也是由人工输入的，其目的是希望看到在看到类似 User 中的句子时，模型应当做出类似 Bot 的回答。

from openai import OpenAI
import json

client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    # base_url="http://localhost:11434/v1"
)

schema = ['日期', '股票名称', '开盘价', '收盘价', '成交量']
examples_data = [  # 示例数据
    {
        "content": "2023-01-10，股市震荡。股票强大科技A股今日开盘价100人民币，一度飙升至105人民币，随后回落至98人民币，最终以102人民币收盘，成交量达到520000。",
        "answers": {
            "日期": "2023-01-10",
            "股票名称": "强大科技A股",
            "开盘价": "100人民币",
            "收盘价": "102人民币",
            "成交量": "520000"
        }
    },
    {
        "content": "2024-05-16，股市利好。股票英伟达美股今日开盘价105美元，一度飙升至109美元，随后回落至100美元，最终以116美元收盘，成交量达到3560000。",
        "answers": {
            "日期": "2024-05-16",
            "股票名称": "英伟达美股",
            "开盘价": "105美元",
            "收盘价": "116美元",
            "成交量": "3560000"
        }
    }
]
questions = [  # 提问问题
    "2025-06-16，股市利好。股票传智教育A股今日开盘价66人民币，一度飙升至70人民币，随后回落至65人民币，最终以68人民币收盘，成交量达到123000。",
    "2025-06-06，股市利好。股票黑马程序员A股今日开盘价200人民币，一度飙升至211人民币，随后回落至201人民币，最终以206人民币收盘。"
]

"""
[
    {"role": "system",      "content": f"你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下示例："},

    {"role": "user",        "content": "2023-01-10，股市震荡。股票强大科技A股今日开盘价100人民币，一度飙升至105人民币，随后回落至98人民币，最终以102人民币收盘，成交量达到520000。"},
    {"role": "assistant",   "content": '{"日期":"2023-01-10","股票名称":"强大科技A股","开盘价":"100人民币","收盘价":"102人民币","成交量":"520000"}'},
    {"role": "user",        "content": "2024-05-16，股市利好。股票英伟达美股今日开盘价105美元，一度飙升至109美元，随后回落至100美元，最终以116美元收盘，成交量达到3560000。"},
    {"role": "assistant",   "content": '{"日期":"2024-05-16","股票名称":"英伟达美股","开盘价":"105美元","收盘价":"116美元","成交量":"3560000"}'},

    {"role": "user",        "content": f"按照上述示例，现在抽取这个句子的信息：{要抽取的句子文本}"}]}
]
"""

messages = [
    {"role": "system", "content": f"你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下示例："},

]

for example in examples_data:
    messages.append(
        {"role": "user", "content":example["content"]}
    )
    messages.append(
        {"role": "assistant", "content":json.dumps(example["content"], ensure_ascii=False)}
    )

for q in questions:
    response = client.chat.completions.create(
        model="qwen3-max",
        messages=messages + [{"role":"user","content":f"按照上述的示例，现在抽取这个句子的信息：{q}"}]
    )
    print(response.choices[0].message.content)

2.6 实战案例（金融文本匹配）

2.6.1 任务介绍

        首先，我们构造几个短文本对：

1. （‘股票市场今日大涨，投资者乐观。’，‘持续上涨的市场让投资者感到满意。’），  
2. （‘油价大幅下跌，能源公司面临挑战。’，‘未来智能城市的建设趋势愈发明显。’），  
3. （‘利率上升，影响房地产市场。’，‘高利率对房地产有一定冲击。’），  

        我们期望模型能够帮我们识别出成对的句子中，2句话是否是有关联的。  
        我们期望模型输出的结果为：  [‘是’，‘不是’，‘是’]

2.6.2 Prompts设计

        在该任务的prompt设计中，我们主要考虑2点：（1）需要向模型解释什么叫作「文本匹配任务」；（2）需要让模型按照我们指定的格式输出。
        为了让模型知道什么叫做「文本匹配任务」，我们借用FewShot的方式，先给模型展示几个正确的例子：

User:
句子一：公司ABC发布了季度财报，显示盈利增长。\n
句子二：财报披露，公司ABC利润上升
Bot:是
User:
句子一：黄金价格下跌，投资者抛售。\n
句子二：外汇市场交易额创下新高
Bot:不是...

        其中，User代表我们输入给模型的句子，Bot代表模型的回复内容。
        注意：上述例子中Bot的部分也是由人工输入的，其目的是希望看到在看到类似User中的句子时，模型应当做出类似Bot的回答。

from openai import OpenAI

client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    # base_url="http://localhost:11434/v1"
)

examples_data = {
    "是": [
        ("公司ABC发布了季度财报，显示盈利增长。", "财报披露，公司ABC利润上升。"),
        ("公司ITCAST发布了年度财报，显示盈利大幅度增长。", "财报披露，公司ITCAST更赚钱了。")
    ],
    "不是": [
        ("黄金价格下跌，投资者抛售。", "外汇市场交易额创下新高。"),
        ("央行降息，刺激经济增长。", "新能源技术的创新。")
    ]
}

questions = [
    ("利率上升，影响房地产市场。", "高利率对房地产有一定的冲击。"),
    ("油价大幅度下跌，能源公司面临挑战。", "未来智能城市的建设趋势越加明显。"),
    ("股票市场今日大涨，投资者乐观。", "持续上涨的市场让投资者感到满意。")
]

"""
    {"role": "system",      "content": f"你帮我完成文本匹配，我给你2个句子，被[]包围，你判断它们是否匹配，回答是或不是，请参考如下示例："},

    {"role": "user",        "content": "句子1：[公司ABC发布了季度财报，显示盈利增长。]句子2：[财报披露，公司ABC利润上升。]"},
    {"role": "assistant",   "content": "是"},
    {"role": "user",        "content": "句子1：[公司ITCAST发布了年度财报，显示盈利大幅度增长。]句子2：[财报披露，公司ITCAST更赚钱了。]"},
    {"role": "assistant",   "content": "是"},
    {"role": "user",        "content": "句子1：[黄金价格下跌，投资者抛售。]句子2：[外汇市场交易额创下新高。]"},
    {"role": "assistant",   "content": "不是"},
    {"role": "user",        "content": "句子1：[央行降息，刺激经济增长。]句子2：[新能源技术的创新。]"},
    {"role": "assistant",   "content": "不是"}, 

    {"role": "user",        "content": f"按照上述示例，回答这2个句子的情况。句子1: [...]，句子2: [...]"}
"""

messages = [
{"role": "system",      "content": f"你帮我完成文本匹配，我给你2个句子，被[]包围，你判断它们是否匹配，回答是或不是，请参考如下示例："},
]

for key, value in examples_data.items():
    for t in value:
        messages.append(
            {"role": "user", "content":f"句子1：[{t[0]}]，句子2：[{t[1]}]"}
        )
        messages.append(
            {"role": "assistant", "content":key}
        )

for q in questions:
    resource = client.chat.completions.create(
        model="qwen3-max",
        messages=messages + [{"role": "user", "content":f"句子1：[{q[0]}]，句子2：[{q[1]}]"}]
    )
    print(resource.choices[0].message.content)

三、RAG开发

3.1 LangChain简介

        LangChain 由Harrison Chase 创建于2022年10月，它是围绕LLMs(大语言模型)建立的一个框架。

        LangChain自身并不开发LLMs，它的核心理念是为各种LLMs实现通用的接口，把LLMs相关的组件“链接”在一起，简化LLMs应用的开发难度，方便开发者快速地开发复杂的LLMs应用。

3.2 LangChain安装

pip install langchain langchain-community langchain-ollama dashscope chromadb

langchain:核心包
langchain-community:社区支持包，提供了更多的第三方模型调用(我们用的阿里云千问模型就需要这个包)
langchain-ollama:Ollama支持包，支持调用Ollama托管部署的本地模型
dashscope:阿里云通义千问的Python SDK
chromadb:轻量向量数据库(后续使用)

3.3 RAG介绍

        通用的基础大模型存在一些问题：（1） LLM的知识不是实时的，模型训练好后不具备自动更新知识的能力，会导致部分信息滞后；（2）LLM领域知识是缺乏的，大模型的知识来源于训练数据，这些数据主要来自公开的互联网和开源数据集，无法覆盖特定领域或高度专业化的内部知识；（3）幻觉问题，LLM有时会在回答中生成看似合理但实际上是错误的信息；（4）数据安全性。

        RAG (Retrieval-Augmented Generation) 即检索增强生成，为大模型提供了从特定数据源检索到的信息，以此来修正和补充生成的答案。可以总结为一个公式：RAG = 检索技术 + LLM提示。

        RAG的工作原理：

        简单来说，RAG工作分为两条线：离线准备线和在线服务线。

        RAG标准流程：

        RAG 标准流程由索引（Indexing）、检索（Retriever）和生成（Generation）三个核心阶段组成。
        （1）索引阶段，通过处理多种来源多种格式的文档提取其中文本，将其切分为标准长度的文本块（chunk），并进行嵌入向量化（embedding），向量存储在向量数据库（vector database）中。
  - 加载文件
  - 内容提取
  - 文本分割，形成chunk
  - 文本向量化
  - 存向量数据库
        （2）检索阶段，用户输入的查询（query）被转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的文本块。
  - query向量化
  - 在文本向量中匹配出与问句向量相似的top_k个
        （3）生成阶段，检索到的相关文本与原始查询共同构成提示词（Prompt），输入大语言模型（LLM），生成精确且具备上下文关联的回答。
  - 匹配出的文本作为上下文和问题一起添加到prompt中
  - 提交给LLM生成答案

        RAG的核心价值：

        （1）解决知识实效性问题：大模型的训练数据有截止时间，RAG 可以接入最新文档（如公司财报、政策文件），让模型输出“与时俱进”。
        （2）降低模型幻觉：模型的回答基于检索到的事实性资料，而非纯靠自身记忆，大幅减少编造信息的概率。
        （3）无需重新训练模型：相比微调（Fine-tuning），RAG 只需更新知识库，成本更低、效率更高。

3.4 向量的基础概念

        向量（Vector）就是文本的“数学身份证”：它把一段文字的语义信息，转换成一串固定长度的数字列表，让计算机能“看懂”文字的含义并做相似度计算。
        简单来说，就是让计算机更方便的理解不同的文本内容，是否表述的是一个意思。

        文本嵌入模型（如text-embedding-v1）通过深度学习等技术，从文本提取语义特征并映射为固定长度的数字序列。

        在向量匹配的过程中，如何识别2段文本是否表述相似的含义，主要可以通过余弦相似度等算法来完成。
        比如（下列案例中向量为示例，仅描述概念，非真实向量）：
- A: “如何快速学习打篮球” ➔ [0.2, 0.5, 0.8]
- B: “打篮球怎么学得快” ➔ [0.18, 0.52, 0.79]
- C: “运动后吃什么好呢” ➔ [0.9, 0.1, 0.2]
        通过余弦相似度算法可以计算得到：A和B相似度0.999789，A和C相似度0.361446
        由此可通过精确的数学计算，去匹配2段文本是否描述同一个意思，提高语义匹配的效率和精度。

        如何更为精准的完成语义匹配，生成向量的维度是一个很重要的指标。如text-embedding-v1模型，可以生成1536维的向量（一段文本固定得到1536个数字序列），比较实用。
• 1536个数字表示，这段文本在1536个主题（抽象的语义特征）方向上的得分（强度）
•生成向量的维度越多，就更好的记录文本的语义特征，做语义匹配会更加精准。
•更多的向量会在计算、存储和匹配过程中，带来更大的压力。
        选择合适的向量维度需要在精确性和性能之间做平衡。 一般1536维算是比较好的选择。

3.5 余弦相似度

        向量的数字序列，共同决定了向量在高维空间中的方向和长度。而余弦相似度主要就是撇除长度的影响，得到方向的夹角。夹角越小越相似，即方向相同。
        如何体现向量的方向和长度呢？以一维向量为例：

        在文本向量语义匹配中，余弦相似度是衡量两个向量方向相似程度的核心算法，即判断两段文本语义是否相近。

        余弦相似度=两个向量的点积÷两个向量模长的乘积

import numpy as np

def get_dot(vec_a, vec_b):
    #计算2个向量的点积，2个向量同维度数字乘积之和
    if len(vec_a) != len(vec_b):
        raise ValueError("2个向量必须维度数量相同")
    dot_sum = 0
    # 遍历每个维度，将对应位置的数值相乘后累加
    for a, b in zip(vec_a, vec_b):
        dot_sum += a * b
    return dot_sum

def get_norm(vec):
    #计算单个向量的模长：对向量的每个数字求平方再求和再开根号
    sum_square = 0
    for v in vec:
        sum_square += v * v
    return np.sqrt(sum_square)

def cosine_similarity(vec_a, vec_b):
    #余弦相似度：2个向量的点积除以2个向量模长的乘积
    result = get_dot(vec_a, vec_b) / (get_norm(vec_a) * get_norm(vec_b))
    return result

if __name__ == '__main__':
    vec_a = [0.5, 0.5]
    vec_b = [0.7, 0.7]
    vec_c = [0.7, 0.5]
    vec_d = [-0.6, -0.5]

    print("ab:", cosine_similarity(vec_a, vec_b))
    print("ac:", cosine_similarity(vec_a, vec_c))
    print("ad:", cosine_similarity(vec_a, vec_d))

3.6 LangChain调用大语言模型

        现在市面上的模型多如牛毛，各种各样的模型不断出现，LangChain模型组件提供了与各种模型的集成，并为所有模型提供一个精简的统一接口。
        LangChain目前支持三种类型的模型：LLMs（大语言模型）、Chat Models（聊天模型）、Embeddings Models（嵌入模型）。
- LLMs：是技术范畴的统称，指基于大参数量、海量文本训练的 Transformer 架构模型，核心能力是理解和生成自然语言，主要服务于文本生成场景。
- 聊天模型：是应用范畴的细分，是专为对话场景优化的 LLMs，核心能力是模拟人类对话的轮次交互，主要服务于聊天场景。
- 文本嵌入模型：文本嵌入模型接收文本作为输入，得到文本的向量。
        LangChain支持的三类模型，它们的使用场景不同，输入和输出不同，开发者需要根据项目需要选择相应。我们所用的阿里云通义千问系列主要来自于：langchain_community包

        阿里云大语言模型的访问：

        LLMs使用场景最多，常用大模型的下载库：
- https://huggingface.co/models
- https://modelscope.cn/models
        同时LangChain支持对许多模型的调用，以通义千问为例：

from langchain_community.llms.tongyi import Tongyi
#实例化模型
llm = Tongyi(model = 'qwen-max') #因为qwen3-max是聊天模型，qwen-max是大语言模型
#模型推理，调用invoke向模型提问
res = llm.invoke(input = "帮我讲个笑话吧")
print(res)

        Ollama本地大语言模型的访问：

        如果要访问本地Ollama的模型，简单更改一下代码。通过langchain_ollama包导入OllamaLLM类即可（请确保Ollama已经启动并提前下载好要使用的模型）。  

from langchain_ollama import OllamaLLM
model = OllamaLLM(model="qwen3:4b")
# 通过invoke方法去调用模型
res = model.invoke(input="你是谁呀能做什么？")
print(res)

3.7 LangChain模型的流式输出

        如果需要流式输出结果，需要将模型的invoke方法改为stream方法即可。（invoke方法是一次性返回完整结果；stream方法是逐段返回结果，流式输出）

from langchain_community.llms.tongyi import Tongyi

model = Tongyi(model="qwen-max")

#通过stream方法获得流式输出
res = model.stream(input="你是谁？")

for chunk in res:
    print(chunk, end="", flush=True)

3.8 LangChain调用聊天模型

        聊天消息包含下面几种类型，使用时需要按照约定传入合适的值：
- AIMessage：就是 AI 输出的消息，可以是针对问题的回答。（OpenAI库中的assistant角色）
- HumanMessage：人类消息就是用户信息，由人给出的信息发送给LLMs的提示信息，比如“实现一个快速排序方法”。（OpenAI库中的user角色）
- SystemMessage：可以用于指定模型具体所处的环境和背景，如角色扮演等。你可以在这里给出具体的指示，比如“作为一个代码专家”，或者“返回json格式”。（OpenAI库中的system角色）

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

#得到模型对象
model = ChatTongyi(model = "qwen3-max")

#准备消息列表
messages = [
    SystemMessage(content="你是一个边塞诗人"),
    HumanMessage(content="写一首唐诗"),
    AIMessage(content="锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
    HumanMessage(content="按照你上一个回复的格式，再写一首唐诗")
]

#调用stream流式执行
res = model.stream(input = messages)

#for循环迭代打印输出，通过.content来获取内容
for chunk in res:
    print(chunk.content , end='', flush=True)

3.9 LangChain消息的简写形式

from langchain_community.chat_models.tongyi import ChatTongyi

#得到模型对象
model = ChatTongyi(model = "qwen3-max")

#准备消息列表
messages = [
    #(角色，内容)   角色：system/human/ai
    ("system", "你是一个边塞诗人"),
    ("human", "写一首唐诗"),
    ("ai", "锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
    ("human", "按照你上一个回复的格式，再写一首唐诗")
]

#调用stream流式执行
res = model.stream(input = messages)

#for循环迭代打印输出，通过.content来获取内容
for chunk in res:
    print(chunk.content , end='', flush=True)

        区别和优势在于，使用类对象的方式，如下：

messages = [
    SystemMessage(content="内容..."),
    HumanMessage(content="内容..."),
    AIMessage(content="内容...")
]

        是静态的，一步到位，直接就得到了Message类的类对象。

        简写形式如下：

messages = [
    ("system", "内容..."),
    ("human", "内容..."),
    ("ai", "内容...")
]

        是动态的，需要在运行时，由LangChain内部机制转换为Message类对象。  

        好处就在于，简写形式避免导包、写起来更简单，更重要的是支持：

messages = [
    ("system", "今天的天气是{weather}"),
    ("human", "我的名字是: {name}"),
    ("ai", "欢迎{lastname}先生"),
]

        由于是动态，需要转换步骤。所以简写形式支持内部填充{变量}占位，可在运行时填充具体值（后续学习提示词模板时用到）

3.10 LangChain调用嵌入模型

        Embedding Models嵌入模型的特点：将字符串作为输入，返回一个浮点数的列表（向量）。在NLP（自然语言处理）中，Embedding的作用就是将数据进行文本向量化。

from langchain_community.embeddings import DashScopeEmbeddings

#创建模型对象，不传model，默认用的是text-embeddings-v1
model = DashScopeEmbeddings()

#用embed_query、embed_documents
print(model.embed_query("我喜欢你"))
print(model.embed_documents(["我喜欢你", "我稀饭你", "晚上吃啥"]))

        模型使用总结：

3.11 LangChain通用提示词模版

        提示词优化在模型应用中非常重要，LangChain提供了PromptTemplate类，用来协助优化提示词。PromptTemplate表示提示词模板，可以构建一个自定义的基础提示词模板，支持变量的注入，最终生成所需的提示词。

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

prompt_template = PromptTemplate.from_template(
    "我的邻居姓{lastname}，刚生了{gender}，请帮我起个名字，简单回答。"
)

#调用.format方法注入信息即可
# prompt_text = prompt_template.format(lastname="黄", gender="女儿")
# 
# model = Tongyi(model = "qwen-max")
# res = model.invoke(input = prompt_text)
# print(res)

model = Tongyi(model="qwen-max")

chain = prompt_template | model
res = chain.invoke(input = {"lastname": "黄", "gender": "女"})
print(res)

3.12 FewShot提示词模版

        FewShotPromptTemplate类对象构建需要5个核心参数：（见如下代码）

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_community.llms.tongyi import Tongyi
#示例的模板
example_template = PromptTemplate.from_template("单词：{word}， 反义词：{antonym}")

#示例的动态数据注入，要求是list内部套字典
example_data = [
    {"word": "大", "antonym": "小"},
    {"word": "上", "antonym": "下"}
]

few_shot_template = FewShotPromptTemplate(
    example_prompt = example_template,                     #示例数据的模板
    examples = example_data,                               #示例的数据（用来注入动态数据的），list内套字典
    prefix = "告知我单词的反义词，我提供如下的示例：",            #示例之前的提示词
    suffix = "基于前面的示例告知我，{input_word}的反义词是？",   #示例之后的提示词
    input_variables = ['input_word']                       #声明在前缀或后缀中所需要注入的变量名
)

prompt_text = few_shot_template.invoke(input = {"input_word": "左"}).to_string()
print(prompt_text)

model = Tongyi(model = "qwen-max")
print(model.invoke(input = prompt_text))

3.13 模板类的format和invoke方法

        PromptTemplate、FewShotPromptTemplate、ChatPromptTemplate都拥有format和invoke这两类方法。

        format和invoke的区别如下：

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate, ChatPromptTemplate

"""
PromptTemplate --> StringPromptTemplate --> BasePromptTemplate
FewShotPromptTemplate --> StringPromptTemplate --> BasePromptTemplate
ChatPromptTemplate --> BaseChatPromptTemplate --> BasePromptTemplate
"""

template = PromptTemplate.from_template("我的朋友是：{lastname}，最喜欢：{hobby}")

res = template.format(lastname="张三", hobby="钓鱼")
print(res, type(res))

res2 = template.invoke({"lastname": "林俊杰", "hobby": "唱歌"})
print(res2, type(res2))

3.14 ChatPromptTemplate的使用

        PromptTemplate：通用提示词模板，支持动态注入信息。

        FewShotPromptTemplate：支持基于模板注入任意数量的示例信息。

        ChatPromptTemplate：支持注入任意数量的历史会话信息。

        通过from_messages方法，从列表中获取多轮次会话作为聊天的基础模板。PromptTemplate类用的from_template仅能接入一条消息，而from_messages可以接入一个list的消息。

        历史会话信息并不是静态的（固定的），而是随着对话的进行不停地积攒，即动态的。所以，历史会话信息需要支持动态注入。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个边塞诗人，可以作诗。"),
        MessagesPlaceholder("history"),  #MessagesPlaceholder作为占位，提供history作为占位的Key
        ("human", "请再来一首唐诗"),
    ]
)

history_data = [
    ("human", "你来写一首唐诗"),
    ("ai", "窗前明月光，疑是地上霜。举头望明月，低头思故乡。"),
    ("human", "好诗，再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
]

prompt_text = chat_prompt_template.invoke({"history": history_data}).to_string()

model = ChatTongyi(model = "qwen3-max")
res = model.invoke(prompt_text)

print(res.content, type(res))   #.content的功能是只取模型回复的内容

3.15 Chain的基础使用

        将组件串联，上一个组件的输出作为下一个组件的输入是 LangChain 链（尤其是 | 管道链）的核心工作原理，这也是链式调用的核心价值：实现数据的自动化流转与组件的协同工作，如下。
                                     chain = prompt_template  |  model
        核心前提：即Runnable子类对象才能入链（以及Callable、Mapping接口子类对象也可加入。我们目前所学习到的组件，均是Runnable接口的子类，如下类的继承关系：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个边塞诗人，可以作诗。"),
        MessagesPlaceholder("history"),  #MessagesPlaceholder作为占位，提供history作为占位的Key
        ("human", "请再来一首唐诗"),
    ]
)

history_data = [
    ("human", "你来写一首唐诗"),
    ("ai", "窗前明月光，疑是地上霜。举头望明月，低头思故乡。"),
    ("human", "好诗，再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
]

model = ChatTongyi(model = "qwen3-max")

#组成链，要求每一个组件都是Runnable接口的子类
chain = chat_prompt_template | model

#通过链去调用invoke或stream
# res = chain.invoke({"history": history_data})
# print(res.content)

#通过stream流式输出
for chunk in chain.stream({ "history": history_data }):
    print(chunk.content, end="", flush=True)

        通过 | 链接提示词模板对象和模型对象，返回值 chain 对象是 RunnableSerializable对象，是 Runnable接口的直接子类  ，也是绝大多数组件的父类 。通过 invoke或stream进行阻塞执行或流式执行。组成的链在执行上有：上一个组件的输出作为下一个组件的输入的特性。  
        所以有如下执行流程：

3.16 或运算符的重写

        前文代码中：  chain = chat_prompt_template | model  
在语法上使用了 | 运算符的重写  

        在 Python 中，运算符（如 +、|）的行为由类的魔法方法决定。例如：  
a + b 本质调用的是 a.__add__(b)  
a | b 本质调用的是 a.__or__(b)  
只需要自行实现类的__or__方法，即可对|符号的功能进行重写。  

        示例：  
- 让 a|b|c 的代码得到一个自定义的类对象(类似列表即[a, b, c])  
- 调用run方法依次输出a、b、c  
- 我们需要重写 | 即 __or__方法

# 定义 Test 类，模拟链中的一个基础组件（比如 LangChain 中的提示词模板或模型）
class Test(object):
    def __init__(self, name):
        # 初始化方法，接收一个 name 参数并保存到实例属性中
        # 这里 name 用于标识这个组件，比如 'a', 'b', 'c'
        self.name = name

    # 魔法方法 __or__ 会在使用 | 操作符时被调用
    # 例如 a | b 实际上就是调用 a.__or__(b)
    # 这是 Python 操作符重载的关键，允许我们自定义 | 的行为
    def __or__(self, other):
        # 当使用 | 连接两个 Test 对象时（如 a | b）
        # 我们不是简单地返回另一个 Test 对象
        # 而是返回一个 MySequence 序列对象，并将 self (当前对象) 和 other (右侧对象) 传入
        # 这一步是“链式调用”的起点：它把两个独立的组件打包成一个序列
        # 相当于 LangChain 中把 prompt 和 model 通过 | 连接成一个 Chain 对象
        return MySequence(self, other)

    # 定义 __str__ 魔法方法，当使用 print() 打印 Test 对象时会被调用
    # 这里返回 self.name，所以 print(a) 会输出 'a'，不加这个方法的话是输出其地址
    def __str__(self):
        return self.name


# 定义 MySequence 类，模拟一个“链”或“序列”容器
# 它的作用是存储多个通过 | 连接起来的组件，并按顺序执行它们
class MySequence(object):
    # 初始化方法，使用 *args 接收任意数量的位置参数
    # *args 会将传入的所有参数打包成一个元组
    def __init__(self, *args):
        # 创建一个空列表，用于按顺序存储链中的各个组件
        self.sequence = []
        # 遍历传入的所有参数（这些参数就是通过 | 逐步添加的组件）
        for arg in args:
            # 将每个组件依次添加到 sequence 列表中
            # 这样 sequence 就维护了组件的执行顺序
            self.sequence.append(arg)

    # 再次定义 __or__ 魔法方法，允许 MySequence 对象继续与右侧组件进行 | 操作
    # 这一步实现了“链式扩展”：chain = a | b | c
    # 实际上执行的是: (a | b) 返回 MySequence(a, b)，然后 (MySequence对象) | c 调用此方法
    def __or__(self, other):
        # 当 MySequence 对象（已有的链）与另一个组件 other 进行 | 操作时
        # 将 other 添加到当前 sequence 列表的末尾
        self.sequence.append(other)
        # 返回 self 自身，实现链式调用
        # 返回 self 很重要，这样连续使用 | 时，每次都是在同一个 MySequence 对象上追加组件
        # 例如 d = a | b | c 的过程：
        # 1. a | b -> 创建 MySequence([a, b])
        # 2. (MySequence对象) | c -> 调用该对象的 __or__(c)，将 c 追加到 sequence 中，返回同一个对象
        return self

    # run 方法模拟链的执行过程
    # 它按照组件添加的顺序，依次“执行”每个组件
    # 在 LangChain 中，这个过程就是 invoke() 或 stream() 做的事情：
    # 将上一个组件的输出作为下一个组件的输入，依次传递数据
    def run(self):
        # 遍历 sequence 列表中的所有组件（按顺序：先添加的先执行）
        for i in self.sequence:
            # 这里简化了执行逻辑，只是打印出每个组件
            # 实际应用中，这里应该是调用组件的某个方法（比如 invoke）
            # 并将上一个组件的输出作为参数传递给下一个组件
            # 例如: output = current_component.invoke(input)
            #       input = output (传递给下一个)
            print(i)  # 由于 Test 类定义了 __str__，所以 print(i) 会输出 i.name


# 程序的主入口，当直接运行这个 .py 文件时，下面的代码会被执行
if __name__ == '__main__':
    # 创建三个 Test 组件实例，分别命名为 'a', 'b', 'c'
    a = Test('a')
    b = Test('b')
    c = Test('c')

    # 使用 | 操作符将三个组件串联成一个链
    # 计算过程如下：
    # 1. 首先计算 a | b
    #    -> 调用 a.__or__(b)
    #    -> 返回 MySequence(a, b)  此时 sequence = [a, b]
    # 2. 然后计算 (a | b) 的结果 | c
    #    -> 调用 MySequence([a,b]).__or__(c)
    #    -> 将 c 追加到 sequence 中，变为 [a, b, c]
    #    -> 返回同一个 MySequence 对象
    # 最终变量 d 就是包含 [a, b, c] 的 MySequence 对象
    d = a | b | c

    # 调用链的 run() 方法，依次执行链中的每个组件
    # 输出结果：
    # a
    # b
    # c
    d.run()

    # 打印 d 的类型，输出: <class '__main__.MySequence'>
    print(type(d))

3.17 Runnable接口

        LangChain 中的绝大多数核心组件都继承了 Runnable 抽象基类（位于 langchain_core.runnables.base）。
                              chain = prompt | model
        chain变量是RunnableSequence（RunnableSerializable子类）类型，而得到这个类型的原因就是Runnable基类内部对__or__魔术方法的改写。同时，在后面继续使用|添加新的组件，依旧会得到RunnableSequence，这就是链的基础架构。

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

prompt = PromptTemplate.from_template("你是一个AI助手")
model = Tongyi(model = "qwen3-max")

chain = prompt | model

print(type(chain))

3.18 StrOutputParser字符串输出解析器

        StrOutputParser是LangChain内置的简单字符串解析器，可以将AIMessage解析为简单的字符串，符合模型invoke方法要求（可传入字符串，不接收AIMessage类型），是Runnable接口的子类（可以加入链）。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import AIMessage

# 创建字符串输出解析器实例
# 作用：后续将模型输出（AIMessage）转换为字符串，便于再次使用
parser = StrOutputParser()
model = ChatTongyi(model = "qwen3-max")
prompt = PromptTemplate.from_template(
    "我的邻居姓：{lastname}，刚生了{gender}，请起名，仅告知我名字无需其他内容。"
)

chain = prompt | model | parser | model | parser

res: str = chain.invoke({"lastname": "黄", "gender": "女儿"})
print(res)
print(type(res))

3.19 JsonOutputParser和多模型执行链

        chain = prompt | model | parser | model | parser
        在前面我们完成了这样的需求去构建多模型链，不过这种做法并不标准，因为：上一个模型的输出，没有被处理就输入下一个模型。正常情况下我们应该有如下处理逻辑：
invoke | stream 初始输入 ➔ 提示词模板 ➔ 模型 ➔ 数据处理 ➔ 提示词模板 ➔ 模型 ➔ 解析器 ➔ 结果
        即：上一个模型的输出结果，应该作为提示词模版的输入，构建下一个提示词，用来二次调用模型。

·      模型的输出为AIMessage类对象，提示词模板要求输入的是字典dict。所以需要完成：将模型输出的AIMessage ➔ 转为字典 ➔ 注入第二个提示词模板中，形成新的提示词（PromptValue对象）。而JsonOutputParser的作用就是将AIMessage转换为一个字典对象。

from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate

#创建所需的解析器
str_parser = StrOutputParser()
json_parser = JsonOutputParser()

#创建模型
model = ChatTongyi(model = "qwen3-max")

#第一个提示词模板
first_prompt = PromptTemplate.from_template(
    "我的邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，仅回复我名字，无需额外内容。"
    "并封装为JSON格式返回给我，要求key是name，value就是你起的名字，请严格遵守格式要求。"
)

#第二个提示词模板
second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)

#构建链
chain = first_prompt | model |json_parser | second_prompt | model |str_parser

for chunk in chain.stream({"lastname": "黄", "gender": "女儿"}):
    print(chunk, end="", flush=True)

小结：

        模型输入：PromptValue或字符串或序列（BaseMessage、list、tuple、str、dict）。
        模型输出：AIMessage
        提示词模板输入：要求是字典
        提示词模板输出：PromptValue对象
        StrOutputParser：AIMessage输入、str输出
        JsonOutputParser：AIMessage输入、dict输出

3.20 自定义函数加入链

chain = first_prompt | model | json_parser | second_prompt | model | str_parser

        前文我们根据JsonOutputParser完成了多模型执行链条的构建。除了JsonOutputParser这类固定功能的解析器之外，我们也可以自己编写Lambda匿名函数来完成自定义逻辑的数据转换，想怎么转换就怎么转换，更自由。想要完成这个功能，可以基于RunnableLambda类实现。

        RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain。语法：RunnableLambda（函数对象或lambda匿名函数）

chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser

        跳过RunnableLambda类，直接让函数加入链也是可以的。因为Runnable接口类在实现 __or__ 的时候，支持Callable接口的实例。函数就是Callable接口的实例

def __or__(self,
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]:

        如上代码示例，| 符号（底层是调用__or__）组链，是支持函数加入的。其本质是将函数自动转换为RunnableLambda。

from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda

model = ChatTongyi(model = "qwen3-max")
str_parser = StrOutputParser()

#第一个提示词模板
first_prompt = PromptTemplate.from_template(
    "我的邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，仅回复我名字，无需额外内容。"
)

#第二个提示词模板
second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)

#函数的入参：AIMessage-->dict ({"name": "xxx"})
my_func = RunnableLambda(lambda ai_msg: {"name":ai_msg.content})

#构建链
chain = first_prompt | model | my_func | second_prompt | model |str_parser

for chunk in chain.stream({"lastname": "黄", "gender": "女儿"}):
    print(chunk, end="", flush=True)

 3.21 Memory临时会话记忆

        如果想要封装历史记录，除了自行维护历史消息外，也可以借助 LangChain 内置的历史记录附加功能。LangChain 提供了 History 功能，帮助模型在有历史记忆的情况下回答。
        基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链（新 Runnable 实例）
        基于InMemoryChatMessageHistory为历史记录提供内存存储（临时用）

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()


def print_prompt(full_prompt):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


base_chain = prompt | print_prompt | model | str_parser


store = {}      # key就是session，value就是InMemoryChatMessageHistory类对象
# 实现通过会话id获取InMemoryChatMessageHistory类对象
def get_history(session_id):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()

    return store[session_id]

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)


if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    # res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    # print("第1次执行：", res)
    #
    # res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    # print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

 3.22 Memory长期会话记忆

        FileChatMessageHistory类实现，核心思路： 基于文件存储会话记录，以session_id为文件名，不同session_id有不同文件存储消息。

        继承BaseChatMessageHistory实现如下3个方法：（1）add_messages:同步模式，添加消息 ；（2）messages:同步模式，获取消息；（3）clear:同步模式，清除消息。

import os, json
from typing import Sequence

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory

# message_to_dict：单个消息对象（BaseMessage类实例） -&gt; 字典
# messages_from_dict：[字典、字典...]  -&gt; [消息、消息...]
# AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类

class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id        # 会话id
        self.storage_path = storage_path    # 不同会话id的存储文件，所在的文件夹路径
        # 完整的文件路径
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 确保文件夹是存在的
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # Sequence序列 类似list、tuple
        all_messages = list(self.messages)      # 已有的消息列表
        all_messages.extend(messages)           # 新的和已有的融合成一个list

        # 将数据同步写入到本地文件中
        # 类对象写入文件 -&gt; 一堆二进制
        # 为了方便，可以将BaseMessage消息转为字典（借助json模块以json字符串写入文件）
        # 官方message_to_dict：单个消息对象（BaseMessage类实例） -&gt; 字典
        # new_messages = []
        # for message in all_messages:
        #     d = message_to_dict(message)
        #     new_messages.append(d)

        new_messages = [message_to_dict(message) for message in all_messages]
        # 将数据写入文件
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(new_messages, f)

    @property       # @property装饰器将messages方法变成成员属性用
    def messages(self) -> list[BaseMessage]:
        # 当前文件内： list[字典]
        try:
            with open(self.file_path, "r", encoding="utf-8") as f:
                messages_data = json.load(f)    # 返回值就是：list[字典]
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)


model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()


def print_prompt(full_prompt):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


base_chain = prompt | print_prompt | model | str_parser

def get_history(session_id):
    return FileChatMessageHistory(session_id, "./chat_history")

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)


if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    # res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    # print("第1次执行：", res)
    #
    # res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    # print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

3.23 CSVLoader

        文档加载器提供了一套标准接口，用于将不同来源（如 CSV、PDF 或 JSON 等）的数据读取为 LangChain 的文档格式。这确保了无论数据来源如何，都能对其进行一致性处理。文档加载器（内置或自行实现）需实现BaseLoader接口。
        Class Document，是 LangChain 内文档的统一载体，所有文档加载器最终返回此类的实例。
一个基础的Document类实例，基于如下代码创建：

from langchain_core.documents import Document

document = Document(
  page_content="Hello, world!", metadata={"source": "https://example.com"}
)

        可以看到，Document类其核心记录了：
page_content：文档内容
metadata：文档元数据（字典）

        不同的文档加载器可能定义了不同的参数，但是其都实现了统一的接口（方法）
- load(): 一次性加载全部文档
- lazy_load(): 延迟流式传输文档，对大型数据集很有用，避免内存溢出。

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path="./data/stu.csv",
    csv_args={
        "delimiter": ",",       # 指定分隔符
        "quotechar": '"',       # 指定带有分隔符文本的引号包围是单引号还是双引号
        # 如果数据原本有表头，就不要下面的代码，如果没有可以使用
        "fieldnames": ['name', 'age', 'gender', '爱好']
    },
    encoding="utf-8"            # 指定编码为UTF-8
)

# 批量加载 .load()   -&gt;  [Document, Document, ...]
# documents = loader.load()
#
# for document in documents:
#     print(type(document), document)

# 懒加载  .lazy_load()  迭代器[Document]
for document in loader.lazy_load():
    print(document)

3.24JSONLoader

        JSONLoader用于将JSON数据加载为Document类型对象。 使用JSONLoader需要额外安装： pip install jq。jq是一个跨平台的json解析工具，LangChain底层对JSON的解析就是基于jq工具实现的。 将JSON数据的信息抽取出来，封装为Document对象，抽取的时候依赖jq_schema语法。

{ "name": "周杰伦",                        • . 表示整个JSON对象（根）
"age": 11,                                 • [] 表示数组       
"hobby": ["唱", "跳", "RAP"],              • .name 表示抽取周杰伦
"other": { "addr": "深圳",                 • .hobby 表示抽取爱好数组
           "tel": "12332112321" }          • .hobby[1] 或 .hobby.[1] 表示抽取跳
}                                          • .other.addr 表示抽取地址深圳

[
{"name": "周杰伦", "age": 11, "gender": "男"},   • .[]. 得到3个字典
{"name": "蔡依临", "age": 12, "gender": "女"},   • .[].name 表示抽取全部的name，即得到3个name信息
{"name": "王力鸿", "age": 11, "gender": "男"}
]

        了解 jq 的基本抽取规则后，即可使用 JSONLoader 加载 JSON 文件了。

from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="xxx.json",     # 文件路径
    jq_schema=".",            # jq schema语法
    text_content=False,       # 抽取的是否是字符串，默认True
    json_lines=True,          # 是否是JsonLines文件（每一行都是JSON的文件）
)

        如下是一个典型的 JsonLines 文件

{"name": "周杰伦", "age": 11, "gender": "男"}
{"name": "蔡依临", "age": 12, "gender": "女"}
{"name": "王力鸿", "age": 11, "gender": "男"}
from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/stu_json_lines.json",
    jq_schema=".name",
    text_content=False,     # 告知JSONLoader 我抽取的内容不是字符串
    json_lines=True         # 告知JSONLoader 这是一个JSONLines文件（每一行都是一个独立的标准JSON）
)

document = loader.load()
print(document)

3.25 TextLoader和文档分割器

        TextLoader的作用是读取文本文件（如.txt），将全部内容放入一个Document对象中。

        RecursiveCharacterTextSplitter，递归字符文本分割器，主要用于按自然段落分割大文档。是 LangChain 官方推荐的默认字符分割器。它在保持上下文完整性和控制片段大小之间实现了良好平衡，开箱即用效果佳。

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# pip install langchain_text_splitters

loader = TextLoader("./data/Python基础语法.txt", encoding="utf-8")

docs = loader.load()        # [Document]

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,         # 分段的最大字符数
    chunk_overlap=50,       # 分段之间允许重叠字符数
    # 文本自然段落分隔的依据符号
    separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
    length_function=len,    # 统计字符的依据函数
)

split_docs = splitter.split_documents(docs)
print(len(split_docs))
for doc in split_docs:
    print("="*20)
    print(doc)
    print("="*20)

3.26 PDFLoader

        LangChain 内支持许多 PDF 的加载器，我们选择其中的 PyPDFLoader 使用。PyPDFLoader 加载器，依赖 PyPDF 库，所以，需要安装它：pip install pypdf。PyPDFLoader 使用还是比较简单的，如下代码即可快速加载 PDF 中的文字内容了：

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(
    file_path="./data/pdf2.pdf",
    mode="single",        # 默认是page模式，每个页面形成一个Document文档对象，
                        # single模式，不管有多少页，只返回1个Document对象
    password="itheima"
)

i = 0
for doc in loader.lazy_load():
    i += 1
    print(doc)
    print("="*20, i)

3.27 Vector Stores向量存储

        基于LangChian的向量库，存储嵌入数据，并执行相似性搜索。

        内存向量存储：
 

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader

vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)


loader = CSVLoader(
    file_path="./data/info.csv",
    encoding="utf-8",
    source_column="source",     # 指定本条数据的来源是哪里
)

documents = loader.load()
# id1 id2 id3 id4 ...
# 向量存储的 新增、删除、检索
vector_store.add_documents(
    documents=documents,        # 被添加的文档，类型：list[Document]
    ids=["id"+str(i) for i in range(1, len(documents)+1)] # 给添加的文档提供id（字符串）  list[str]
)

# 删除  传入[id, id...]
vector_store.delete(["id1", "id2"])

# 检索 返回类型list[Document]
result = vector_store.similarity_search(
    "瑞达法",
    3       # 检索的结果要几个
)

print(result)

        外部向量持久化存储：

from langchain_chroma import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader

# Chroma 向量数据库（轻量级的）
# 确保 langchain-chroma chromadb 这两个库安装了的，没有的话请pip install

vector_store = Chroma(
    collection_name="test",     # 当前向量存储起个名字，类似数据库的表名称
    embedding_function=DashScopeEmbeddings(),       # 嵌入模型
    persist_directory="./chroma_db"     # 指定数据存放的文件夹
)


# loader = CSVLoader(
#     file_path="./data/info.csv",
#     encoding="utf-8",
#     source_column="source",     # 指定本条数据的来源是哪里
# )
#
# documents = loader.load()
# # id1 id2 id3 id4 ...
# # 向量存储的 新增、删除、检索
# vector_store.add_documents(
#     documents=documents,        # 被添加的文档，类型：list[Document]
#     ids=["id"+str(i) for i in range(1, len(documents)+1)] # 给添加的文档提供id（字符串）  list[str]
# )
#
# # 删除  传入[id, id...]
# vector_store.delete(["id1", "id2"])

# 检索 返回类型list[Document]
result = vector_store.similarity_search(
    "Python是不是简单易学呀",
    3,        # 检索的结果要几个
    filter={"source": "黑马程序员"}
)

print(result)

3.28 基于向量检索构建提示词

        向量存储的实例，通过add_texts(list[str])方法可以快速添加到向量存储中。

        流程：1.先通过向量存储检索匹配信息；2.将用户提问和匹配信息一同封装到提示词模板中提问模型。

"""
提示词：用户的提问 + 向量库中检索到的参考资料
"""
from langchain_community.chat_models import ChatTongyi
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

def print_prompt(prompt):
    print(prompt.to_string())
    print("=" * 20)
    return prompt

model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问：{input}")
    ]
)

vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))

# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]
vector_store.add_texts(
    ["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

# 检索向量库
result = vector_store.similarity_search(input_text, 2)
reference_text = "["
for doc in result:
    reference_text += doc.page_content
reference_text += "]"

chain = prompt | print_prompt | model | StrOutputParser()

res = chain.invoke({"input": input_text, "context": reference_text})
print(res)

3.29 RunnablePassthrough的使用

        作用是让向量检索加入链。

"""
提示词：用户的提问 + 向量库中检索到的参考资料
"""
from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser


def print_prompt(prompt):
    print(prompt.to_string())
    print("=" * 20)
    return prompt


model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问：{input}")
    ]
)

vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))

# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]
vector_store.add_texts(
    ["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

# langchain中向量存储对象，有一个方法：as_retriever，可以返回一个Runnable接口的子类实例对象
retriever = vector_store.as_retriever(search_kwargs={"k": 2})


def format_func(docs: list[Document]):
    if not docs:
        return "无相关参考资料"

    formatted_str = "["
    for doc in docs:
        formatted_str += doc.page_content
    formatted_str += "]"

    return formatted_str

# chain
chain = (
    {"input": RunnablePassthrough(), "context": retriever | format_func} | prompt | print_prompt | model | StrOutputParser()
)

res = chain.invoke(input_text)
print(res)
"""
retriever:
    - 输入：用户的提问       str
    - 输出：向量库的检索结果  list[Document]
prompt:
    - 输入：用户的提问 + 向量库的检索结果   dict
    - 输出：完整的提示词                 PromptValue
"""

四、Agent智能体

4.1 Agent智能体介绍

        智能体（Agent） 是一种能够自主规划、决策、执行任务的组件，核心是让大语言模型（LLM）根据任务需求，选择并调用工具，完成单靠模型自身无法解决的复杂问题。没有 Agent 时，LLM 只能基于自身训练数据回答问题，遇到需要实时数据、复杂计算、外部工具调用的场景就会卡壳。有了 Agent 后，LLM 就像一个 "指挥官"，能思考任务步骤→选择合适工具→执行工具调用→根据结果调整策略，直到完成任务。

        核心特点：
- 目标驱动：围绕用户的具体任务目标展开工作。
- 工具调用能力：能连接外部工具，弥补 LLM 的局限性。
- 自主决策与迭代：不需要人工干预，能根据工具返回的结果，判断是否需要继续调用工具，或直接生成最终答案。

        以电商商品问答为例：

4.2 Agent智能体初体验

        设计一个天气查询工具，让模型用于天气回答的能力。

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="查询天气")
def get_weather() -> str:
    return "晴天"

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),        # 智能体的大脑LLM
    tools=[get_weather],            # 向智能体提供工具列表
    system_prompt="你是一个聊天助手，可以回答用户问题。",
)

res = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "明天深圳的天气如何？"},
        ]
    }
)

for msg in res["messages"]:
    print(type(msg).__name__, msg.content)

4.3 Agent的流式输出

        通过create_agent方法可以创建Agent对象，其也是Runnable接口的子类实现，所以也拥有：
- invoke，执行，一次型得到完整结果
- stream，执行，流式得到结果

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="获取股价，传入股票名称，返回字符串信息")
def get_price(name: str) -> str:
    return f"股票{name}的价格是20元"

@tool(description="获取股票信息，传入股票名称，返回字符串信息")
def get_info(name: str) -> str:
    return f"股票{name}，是一家A股上市公司，专注于IT职业教育。"

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_price, get_info],
    system_prompt="你是一个智能助手，可以回答股票相关问题，记住请告知我思考过程，让我知道你为什么调用某个工具"
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "传智教育股价多少，并介绍一下"}]},
    stream_mode="values"
):
    latest_message = chunk['messages'][-1]

    if latest_message.content:
        print(type(latest_message).__name__, latest_message.content)

    try:
        if latest_message.tool_calls:
            print(f"工具调用： { [tc['name'] for tc in latest_message.tool_calls]  }")
    except AttributeError as e:
        pass

4.4 LangChain的ReAct行动框架

        Agent ReAct 是大模型智能体的核心思考与行动框架，全称 Reasoning + Acting（推理 + 行动），是让 Agent 像人类一样「思考问题→制定策略→执行行动→验证结果」的关键逻辑。

        简单来说：ReAct 让 Agent 不再是 “直接回答问题”，而是通过 “自然语言思考过程” 指导工具调用，一步步解决复杂问题，完美适配需要多步推理、工具协作的场景（如智能客服、报告生成、任务规划等）。

        一个典型的 ReAct 范式的 Agent 如图所示：
- 思考 Reasoning：分析问题，判断现有信息是否足够，明确下一步。即模型决策是否需要调用外部工具获取更多信息用来回答。
- 行动 Action：执行思考阶段指定的策略。即基于模型决策结果，调用工具获取信息
- 观察 Observation：获取行动的结果，提取有效信息。即获取工具返回值即判断工具是否正常工作，为下一轮思考提供信息。
- （再）思考 → （再）行动 → （再）观察 → 循环往复直到结束。

        LangChain 的 Agent 对象遵循 ReAct 框架要求，在执行的过程中会持续的自我思考、自我行动、自我观察。

        一个典型的 ReAct 案例如下：

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="获取体重，返回值是整数，单位千克")
def get_weight() -> int:
    return 90

@tool(description="获取身高，返回值是整数，单位厘米")
def get_height() -> int:
    return 172

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weight, get_height],
    system_prompt="""你是严格遵循ReAct框架的智能体，必须按「思考→行动→观察→再思考」的流程解决问题，
    且**每轮仅能思考并调用1个工具**，禁止单次调用多个工具。
    并告知我你的思考过程，工具的调用原因，按思考、行动、观察三个结构告知我""",
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "计算我的BMI"}]},
    stream_mode="values"
):
    latest_message = chunk['messages'][-1]

    if latest_message.content:
        print(type(latest_message).__name__, latest_message.content)

    try:
        if latest_message.tool_calls:
            print(f"工具调用： { [tc['name'] for tc in latest_message.tool_calls]  }")
    except AttributeError as e:
        pass

4.5 Agent的middleware中间件

        中间件的作用是对智能体的每一步工作进行控制和自定义的执行。

        作用场景：
- 日志记录、分析、调试
- 转换提示词、工具选择
- 重试、备用、提前终止等逻辑控制
- 安全防护、个人身份检测等

 

        LangChain 中内置了一些基础的中间件，参见：https://docs.langchain.com/oss/python/langchain/middleware/built-in。中间件通过 Hooks 钩子来实现拦截，自定义中间件可以简单的使用装饰器来定义。
        节点式钩子（执行点顺序拦截）：
before_agent: agent 执行之前拦截
after_agent: agent 执行后拦截
before_model: 模型执行前拦截
after_model: 模型执行后拦截
        针对工具和模型的包装式钩子：
wrap_model_call: 每个模型调用时候拦截
wrap_tool_call: 每个工具调用时候拦截

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_agent, after_agent, before_model, after_model, wrap_model_call, \
    wrap_tool_call
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool
from langgraph.runtime import Runtime

@tool(description="查询天气，传入城市名称字符串，返回字符串天气信息")
def get_weather(city: str) -> str:
    return f"{city}天气：晴天"

"""
1. agent执行前
2. agent执行后
3. model执行前
4. model执行后
5. 工具执行中
6. 模型执行中
"""

@before_agent
def log_before_agent(state: AgentState, runtime: Runtime) -&gt; None:
    # agent执行前会调用这个函数并传入state和runtime两个对象
    print(f"[before agent]agent启动，并附带{len(state['messages'])}消息")

@after_agent
def log_after_agent(state: AgentState, runtime: Runtime) -&gt; None:
    print(f"[after agent]agent结束，并附带{len(state['messages'])}消息")

@before_model
def log_before_model(state: AgentState, runtime: Runtime) -&gt; None:
    print(f"[before_model]模型即将调用，并附带{len(state['messages'])}消息")

@after_model
def log_after_model(state: AgentState, runtime: Runtime) -&gt; None:
    print(f"[after_model]模型调用结束，并附带{len(state['messages'])}消息")

@wrap_model_call
def model_call_hook(request, handler):
    print("模型调用啦")
    return handler(request)

@wrap_tool_call
def monitor_tool(request, handler):
    print(f"工具执行：{request.tool_call['name']}")
    print(f"工具执行传入参数：{request.tool_call['args']}")

    return handler(request)

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weather],
    middleware=[log_before_agent, log_after_agent, log_before_model, log_after_model, model_call_hook, monitor_tool]
)

res = agent.invoke({"messages": [{"role": "user", "content": "深圳今天的天气如何呀，如何穿衣"}]})
print("**********\n", res)

# RAG
Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

cover

大模型安全实战】智能体安全的攻防二元性：同一种能力，为什么既是防线也是攻击面？（第4期）

avatar 龙虾开发者社区
cover

SparkSQL 之 Hive On Spark 原理分析

avatar 龙虾开发者社区

2026大厂AI Agent面试深度复盘：从“会背概念“到“能扛生产“的范式跃迁

2026年AI Agent面试核心要点摘要（147字） 大厂AI Agent面试已从概念背诵转向工程实践考核，重点考察：1）理解Agent与Chatbot的本质差异在于闭环行动能力；2）掌握ReAct循环、记忆分层设计（工作/短期/长期记忆）及优化策略；3）熟悉MCP协议对工具调用的标准化改进及安全风险；4）具备异常处理、性能优化的工程能力。面试官更关注候选人在非确定性系统中构建确定性保障的能力，

avatar 龙虾开发者社区