一、基础

01 阿里云百炼Key使用

from openai import OpenAI
import os

client = OpenAI(
    # 如果没有配置环境变量，请用阿里云百炼API Key替换：api_key="sk-xxx"
    # api_key="sk-8e23095db89341679edc0cfa52b8a53",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

messages = [{"role": "user", "content": "你是谁"}]
completion = client.chat.completions.create(
    model="deepseek-v3",  # 您可以按需更换为其它深度思考模型
    messages=messages,
    extra_body={"enable_thinking": True},
    stream=True
)
is_answering = False  # 是否进入回复阶段
print("\n" + "=" * 20 + "思考过程" + "=" * 20)
for chunk in completion:
    delta = chunk.choices[0].delta
    if hasattr(delta, "reasoning_content") and delta.reasoning_content is not None:
        if not is_answering:
            print(delta.reasoning_content, end="", flush=True)
    if hasattr(delta, "content") and delta.content:
        if not is_answering:
            print("\n" + "=" * 20 + "完整回复" + "=" * 20)
            is_answering = True
        print(delta.content, end="", flush=True)

02 OpenAI使用

from openai import OpenAI

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

    )

response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role": "user", "content": "你是python专家，并且不说废话"},
        {"role": "assistant", "content": "是的，我是一个python专家"},
        {"role": "user", "content": "那么，你能不能帮我写一个python程序，计算1到100的和"},
    ]
)


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

03 OpenAI库流式调用

from openai import OpenAI

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

)

response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role": "user", "content": "你是python专家，并且不说废话"},
        {"role": "assistant", "content": "是的，我是一个python专家"},
        {"role": "user", "content": "那么，你能不能帮我写一个python程序，计算1到100的和"},
    ],
    stream=True
)

for chunk in response:
    print(chunk.choices[0].delta.content,
          end=" ",
          flush=True)

04 附带历史消息

from openai import OpenAI

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

)

response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role": "system", "content": "你是AI助理，回答很简洁"},
        {"role": "user", "content": "小明有两条宠物"},
        {"role": "assistant", "content": "好的"},
        {"role": "user", "content": "小红有三条宠物"},
        {"role": "assistant", "content": "好的"},
        {"role": "user", "content": "总共有几只宠物"},
    ],
    stream=True
)

for chunk in response:
    print(chunk.choices[0].delta.content,
          end=" ",
          flush=True)

05  Prompt提示词案例-金融文本分类

from openai import OpenAI

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

)

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

# 分类列表
examples_types = ['新闻报道', '财务报道', '公司公告', '分析师报告']

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

"""
[
    {"role": "system",     "content": "你是金融专家，将文本分类为['新闻报道', '财务报道', '公司公告', '分析师报告']，不清楚的分类为'不清楚'"},
    {"role": "user",       "content": "今日，央行发布公告宣布................"},
    {"role": "assistant",  "content": "新闻报道"},
    {"role": "user",       "content": "ABC公司今日发布公告称，已成功完成对XYZ公司股................"},
    {"role": "assistant",  "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",
        messages=messages + [{"role": "user", "content": f"按照示例，回答这段文本的分类类别: {q}"}]
    )
    print(response.choices[0].message.content)

06 提示词优化-金融信息抽取

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"}'},
    ...
]
"""

messages = [
    {"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"}'},
]

for example in examples_data:
    messages.append(
        {"role": "user", "content": example["content"]}
    )

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

# for x in messages:
#     print(x)
'''
{"日期": "2025-06-16", "股票名称": "传智教育A股", "开盘价": "66人民币", "收盘价": "68人民币", "成交量": "123000"}
{"日期": "2025-06-06", "股票名称": "黑马程序员A股", "开盘价": "200人民币", "收盘价": "206人民币", "成交量": "原文未提及"}
'''

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)

07 提示词优化，金融文本匹配判断

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:
    response = client.chat.completions.create(
            model="qwen3-max",
            messages=messages+[
                {"role": "user", "content": f"句子1：[{q[0]}]句子2：[{q[1]}]"}
            ]
        )
    print(response.choices[0].message.content)

二、RAG开发

01 LangChain介绍

开发LLM相关业务功能的集成，python第三方库，提供各种接口

02 安装环境

pip install langchain langchain-community langchain-ollama dashscope chromadb

各包作用说明

• langchain：核心包，LangChain 框架的基础功能模块。
• langchain-community：社区支持包，提供更多第三方模型的调用能力（如阿里云通义千问模型）。
• langchain-ollama：Ollama 支持包，用于调用 Ollama 托管部署的本地大语言模型。
• dashscope：阿里云通义千问的 Python SDK，用于直接调用阿里云模型服务。
• chromadb：轻量级向量数据库，用于后续 RAG 等场景的向量存储与检索。

03 RAG检索增强

通用基础大模型存在的问题

• 知识非实时性：LLM 训练完成后不具备自动更新知识的能力，会导致部分信息滞后。
• 领域知识匮乏：大模型知识主要来源于公开互联网和开源数据集，无法覆盖特定领域或高度专业化的内部知识。
• 幻觉问题：LLM 有时会生成看似合理但实际上错误的信息。
• 数据安全性：（图片中该点未展示完整内容）

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

RAG=检索技术+LLM 提示

RAG工作分为两条线：

• 离线准备线
• 在线服务线

04 扩展向量Vector

• 向量（Vector）就是文本的 “数学身份证”：它把一段文字的语义信息，转换成一串固定长度的数字列表，让计算机能 “看懂” 文字的含义并做相似度计算。
• 文本嵌入模型embedding - 深度学习技术
• 维度越多，语义匹配越精准；但性能压力也会增大

05 余弦相似度算法

import numpy as np

"""
计算两个向量的余弦相似度（衡量方向相似性，剔除长度影响）

参数：
    vec_a (np.array): 向量A
    vec_b (np.array): 向量B
返回：
    float: 余弦相似度结果（范围[-1,1]，越接近1方向越一致）
公式：
    cos_sim = (vec_a · vec_b) / (||vec_a|| × ||vec_b||)
拆解：
    1. 点积：vec_a · vec_b = vec_a[0]×vec_b[0] + vec_a[1]×vec_b[1] + ... + vec_a[n]×vec_b[n]
    2. 模长：||vec_a|| = √(vec_a[0]² + vec_a[1]² + ... + vec_a[n]²)
    3. 模长：||vec_b|| = √(vec_b[0]² + vec_b[1]² + ... + vec_b[n]²)

A: [0.5, 0.5]
B: [0.7, 0.7]
C: [0.7, 0.5]
D: [-0.6, -0.5]
"""

def get_dot(vec_a, vec_b):
    """计算2个向量的点积，2个向量同维度"""
    if len(vec_a) != len(vec_b):
        raise ValueError

    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

    # numpy sqrt函数完成开根号
    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))

06 LangChain调用大语言模型

LangChain 模型支持分类

现在市面上的模型种类繁多，LangChain 模型组件提供了与各种模型的集成，并为所有模型提供一个精简的统一接口。
LangChain 目前支持三种类型的模型：LLMs（大语言模型）、Chat Models（聊天模型）、Embeddings Models（嵌入模型）。

1. LLMs（大语言模型）：是技术范畴的统称，指基于大参数数量、海量文本训练的 Transformer 架构模型，核心能力是理解和生成自然语言，主要服务于文本生成场景。
2. Chat Models（聊天模型）：是应用范畴的细分，是专为对话场景优化的 LLMs，核心能力是模拟人类对话的轮次交互，主要服务于聊天场景。
3. 文本嵌入模型（Embeddings Models）：接收文本作为输入，将文本转换为向量（Embedding），主要用于语义表示、相似度计算、检索等 RAG 相关场景。

LangChain 支持的三类模型，它们的使用场景不同，输入和输出不同，开发者需要根据项目需要选择相应。

我们所用的阿里云通义千问系列主要来自于：langchain_community 包。

from langchain_community.llms.tongyi import Tongyi

# 不用qwen3-max， 因为qwen3-max是聊天模型， qwen-max是大语言模型
model = Tongyi(model="qwen-max")

# 调用invoke向模型提问
res = model.invoke(input="你是谁啊能做什么？")
print(res)


# from langchain_ollama import OllamaLLM
'''本地测试'''
# model = OllamaLLM(model="qwen2.5:14b")
# res = model.invoke(input="你是谁啊能做什么？")
# print(res)

res = model.stream(input="你是谁呀能做什么？")
for chunk in res:
    print(chunk, end="", flush=True)

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="按照你上一个回复的格式，在写一首唐诗")
]

res = model.stream(input=messages)

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

09 消息的简写

messages = [
    ("system", "你是一个边塞诗人"),
    ("human", "写一首唐诗"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦。"),
    ("system", "按照你上一个回复的格式，在写一首唐诗"),
]

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

好处：简写形式避免导包、写起来更简单、简写形式支持内部填充{变量}占位、可在运行时填充具体值。

10 调用文本嵌入模型

Embeddings Models（嵌入模型）的特点：将字符串作为输入，返回一个浮点数的列表（向量）。

在 NLP（自然语言处理）中，Embedding（嵌入） 的核心作用就是将文本数据进行向量化，把人类可读的文字转换为计算机可计算的向量形式，从而实现语义相似度计算、检索等功能。

from langchain_community.embeddings import DashScopeEmbeddings

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

print(model.embed_query("我喜欢你"))
print(model.embed_documents(["我喜欢你", "你真聪明"]))

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

prompt_template = PromptTemplate.from_template(
    "我的邻居姓{lastname}, 刚生了{gender}, 你帮我起个名字， 简单回答。"
)
model = Tongyi(model="qwen-max")

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

chain = prompt_template | model

res = chain.invoke(input={"lastname": "张", "gender": "女儿"})
print(res)

rom langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
from langchain_community.llms.tongyi import Tongyi

# 示例模板
example_template = PromptTemplate.from_template("单词: {word}，反义词：{antonym}")

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

few_shot_template = FewShotPromptTemplate(
    example_prompt=example_template,  # 示例数据的模版
    examples=examples_data,      # 示例的数据
    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))

from langchain_core.prompts import PromptTemplate
from langchain_core.prompts import FewShotPromptTemplate
from langchain_core.prompts import ChatPromptTemplate

template = PromptTemplate.from_template("我的邻居是:{lastname}, 最喜欢： {hobby}")

res = template.format(lastname="Rudy", hobby=" running")
print(res, type(res))

res2 = template.invoke({"lastname": "Rudy", "hobby": " running"})
print(res2, type(res2))

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

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个 helpful assistant"),
        MessagesPlaceholder("history"),
        ('human', "请再来一首唐诗"),
    ]
)

history_data = [
    ("human", "你来写一首唐诗"),
    ("ai", "床前明月光，疑是地上霜。"),
    ("human", "再来一句"),
    ("ai", "白日依山，黄河入海。"),
]

prompt_text = chat_prompt_template.invoke({"history": history_data})
model = ChatTongyi(model="qwen3-max")
res = model.invoke(prompt_text)
print(res)