LangChain 入门与 Model I/O

# -*- coding: utf-8 -*-
"""
三类模型：对话模型、文本续写模型、嵌入模型
==========================================
它们的差别不在「聪不聪明」，而在输入输出的形状不同，
因此在流水线里站的工位也不同。
"""

import os

from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain_core.messages import HumanMessage, SystemMessage

if not os.environ.get("OPENAI_API_KEY"):
    raise SystemExit("请先设置环境变量 OPENAI_API_KEY")

# ---------------------------------------------------------------------------
# 一、对话模型（ChatModel）：进消息列表，出消息对象
# 日常开发的绝对主力，本讲后面所有例子都用它。
# ---------------------------------------------------------------------------
chat = init_chat_model("openai:gpt-4o-mini")

messages = [
    SystemMessage(content="你是一位擅长人工智能学科的专家，回答尽量简短"),
    HumanMessage(content="请解释一下什么是机器学习"),
]
reply = chat.invoke(messages)

print(type(reply))        # AIMessage
print(reply.content)      # 文本正文
print(reply.response_metadata.get("finish_reason"))   # 结束原因

# 多轮的本质：把上一轮的 AIMessage 接回列表，再追加新的 HumanMessage。
messages.append(reply)
messages.append(HumanMessage(content="用一个比喻再讲一遍"))
print(chat.invoke(messages).content)

# ---------------------------------------------------------------------------
# 二、文本续写模型（LLM / Text Model）：进字符串，出字符串
# 没有角色概念，也不维护对话结构，只做「续写」。
# 现在只在少数老接口上还能见到，新项目一律用对话模型。
# ---------------------------------------------------------------------------
#   from langchain_openai import OpenAI
#   completion = OpenAI()                      # 需要提供方仍然开放这类老接口
#   text = completion.invoke("写一首关于春天的诗")
#   print(type(text))                          # <class 'str'>
#
# 判断标准很简单：
#   需要设定角色、需要多轮、需要工具调用 —— 对话模型；
#   只要单次续写一段文字      —— 续写模型也够用，但没必要。

# ---------------------------------------------------------------------------
# 三、嵌入模型（Embedding Model）：进文本，出一串浮点数
# 它不生成任何自然语言，产出的是可以做相似度计算的向量。
# 检索、聚类、去重、推荐都靠它，属于另一条技术线。
# ---------------------------------------------------------------------------
emb = init_embeddings("openai:text-embedding-3-small")

one = emb.embed_query("这是第一个测试文档")
print(len(one), one[:5])          # 向量维度，以及前五个分量

many = emb.embed_documents(["这是第一个测试文档", "这是第二个测试文档"])
print(len(many), len(many[0]))    # 两条文本，各自一个等长向量

# embed_query 与 embed_documents 的区别不只是参数个数：
# 部分提供方会对「查询」和「被检索的文档」用不同的处理方式，
# 所以别用一个方法硬顶两种场景。

# -*- coding: utf-8 -*-
"""
消息类型：对话模型真正吃进去的东西
====================================
对话模型的输入不是一段文本，而是一串带角色的消息。
角色决定了模型怎么理解这句话：是规矩、是提问，还是它自己说过的话。
"""

from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

model = init_chat_model("openai:gpt-4o-mini")

# ---------------------------------------------------------------------------
# 三种最常用的消息
# ---------------------------------------------------------------------------
# SystemMessage：定规矩。人设、语气、输出格式要求都写这里，
#                通常放在列表第一条，且整轮对话保持不变。
# HumanMessage ：用户说的话。
# AIMessage    ：模型说过的话。多轮对话时由你把它接回列表。
messages = [
    SystemMessage(content="你是一位著名的诗人，回答只写诗，不写解释"),
    HumanMessage(content="给我写一首唐诗"),
]

first = model.invoke(messages)
print(first.content)

# 把模型这一轮的回答接回去，再追加新问题，就是第二轮。
messages.append(first)
messages.append(HumanMessage(content="再写一首宋词"))
second = model.invoke(messages)
print(second.content)

# 此时 messages 已经有 4 条。模型能接上第二个问题，
# 不是因为它「记住了」，而是因为整段历史又被完整发了一遍。

# ---------------------------------------------------------------------------
# 另外两种消息，用到时再认
# ---------------------------------------------------------------------------
# ChatMessage：自定义角色，少见。
# ToolMessage：工具执行结果的回执，出现在工具调用场景里，
#              协议层的细节在 Function Call 那一讲。

# ---------------------------------------------------------------------------
# 简写形式：元组或字典
# ---------------------------------------------------------------------------
# 下面三种写法等价，选一种团队内统一即可。
model.invoke([SystemMessage(content="你是翻译助手"), HumanMessage(content="今天天气不错")])
model.invoke([("system", "你是翻译助手"), ("human", "今天天气不错")])
model.invoke([{"role": "system", "content": "你是翻译助手"},
              {"role": "user", "content": "今天天气不错"}])

# ---------------------------------------------------------------------------
# 标准内容块：跨提供方读同一份结构
# ---------------------------------------------------------------------------
# 不同家的响应体字段五花八门，1.x 用 content_blocks 把它们统一成一种结构，
# 推理过程、正文、工具调用请求都能用同一套代码取出来。
resp = model.invoke("简单算一下 1 加 2")
for block in resp.content_blocks:
    print(block["type"])          # 例如 "text"

# content 仍然可以直接当字符串用，简单场景不必动 content_blocks。
print(resp.content)

# -*- coding: utf-8 -*-
"""
模型参数与 Token：控制成本与稳定性的两个旋钮
==============================================
参数调不好，表现出来就是「一会儿答得挺好，一会儿胡说八道」，
或者「答到一半突然断了」。这两种现象各有各的原因。
"""

from langchain.chat_models import init_chat_model

# ---------------------------------------------------------------------------
# temperature：随机性。取值越大越发散，越小越稳定。
# ---------------------------------------------------------------------------
# 需要结果可复现、能被下游程序解析（抽字段、写 SQL、做分类）时压到 0；
# 写文案、起名字、头脑风暴这类要多样性的场景才调高。
strict = init_chat_model("openai:gpt-4o-mini", temperature=0)
creative = init_chat_model("openai:gpt-4o-mini", temperature=0.9)

question = "给一家卖手冲咖啡的小店起个名字"
print("稳定档：", strict.invoke(question).content)
print("发散档：", creative.invoke(question).content)

# 同一个问题连问两次，temperature=0 的结果基本一致，0.9 的每次都不一样。

# ---------------------------------------------------------------------------
# max_tokens：这次最多生成多少 token。它截的是「输出」，不是「输入」。
# ---------------------------------------------------------------------------
short = init_chat_model("openai:gpt-4o-mini", max_tokens=20)
reply = short.invoke("详细介绍一下什么是大模型")
print(reply.content)
# 输出会在第 20 个 token 处硬生生断掉，句子可能只说了一半。
print(reply.response_metadata.get("finish_reason"))   # length 表示被截断

# 结论：句子莫名其妙断掉，先看 finish_reason 是不是 length，
# 而不是去怀疑模型「变笨了」。

# ---------------------------------------------------------------------------
# Token 是什么
# ---------------------------------------------------------------------------
# 模型处理文本的最小单位，生成时一个接一个往外吐，
# 每个新 token 都基于前面全部 token 预测出来——这也是流式输出能逐字显示的原因。
#
# 粗略换算：1 个中文 token 约等于 1 到 1.8 个汉字，
#          1 个英文 token 约等于 3 到 4 个字母。
#
# 计费与上下文窗口都按 token 算，所以历史越长、单次越贵。

usage = strict.invoke("你好").usage_metadata
print(usage["input_tokens"], usage["output_tokens"], usage["total_tokens"])

# 想在发请求前就估算长度，可以直接问模型对象：
print(strict.get_num_tokens("这是一句用来数 token 的中文"))

# -*- coding: utf-8 -*-
"""
四种调用方式：invoke / batch / stream / 异步
=============================================
LangChain 里的每个组件——提示词模板、模型、解析器——都实现了同一套
Runnable 协议，所以这四个方法在任何组件上的用法完全一样。
这也是下一讲能用一个管道符把它们串起来的前提。
"""

import asyncio

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

model = init_chat_model("openai:gpt-4o-mini")

# ---------------------------------------------------------------------------
# 一、invoke：处理单条输入，等模型全部推理完再一次性返回
# ---------------------------------------------------------------------------
reply = model.invoke("请介绍一下你自己")
print(reply.content)
# 体验上像「提交请求，等待结果」：安静几秒，然后整段文字突然出现。
# 做摘要、抽字段、分类这类后台任务，这种方式最简单也最稳。

# ---------------------------------------------------------------------------
# 二、stream：流式，一个 token 一个 token 地往外吐
# ---------------------------------------------------------------------------
print("开始流式输出：")
for chunk in model.stream("写一首关于春天的短诗"):
    print(chunk.content, end="", flush=True)   # flush 保证立刻显示，不进缓冲区
print("\n流式输出结束")
# 每个 chunk 是一小块增量，不是完整答案。
# 聊天界面必须用它，否则用户会盯着空白等十几秒。

# 需要把流式结果同时展示并留存时，一边打印一边拼：
full = ""
for chunk in model.stream("用两句话解释什么是提示词"):
    full += chunk.content
    print(chunk.content, end="", flush=True)
print()
print("完整文本长度：", len(full))

# ---------------------------------------------------------------------------
# 三、batch：一次提交多条互不相关的输入，并发跑
# ---------------------------------------------------------------------------
tasks = [
    [SystemMessage(content="你是一位乐于助人的小助手"), HumanMessage(content="什么是机器学习")],
    [SystemMessage(content="你是一位乐于助人的小助手"), HumanMessage(content="什么是深度学习")],
    [SystemMessage(content="你是一位乐于助人的小助手"), HumanMessage(content="什么是强化学习")],
]
results = model.batch(tasks)
for item in results:
    print(item.content[:40], "...")

# 返回结果的顺序与输入顺序严格一致，可以直接按下标取。
# 注意：batch 是并发不是排队，三条同时打出去，小心提供方的频率限制。
# 需要限制并发数时传配置：
results = model.batch(tasks, config={"max_concurrency": 2})

# ---------------------------------------------------------------------------
# 四、异步版本：ainvoke / abatch / astream
# ---------------------------------------------------------------------------
async def main():
    one = await model.ainvoke("一句话介绍长城")
    print(one.content)

    async for chunk in model.astream("一句话介绍黄河"):
        print(chunk.content, end="", flush=True)
    print()

asyncio.run(main())

# 选择很直白：
#   后台批处理     -> batch
#   面向用户的界面 -> stream
#   服务端高并发   -> 异步版本
#   其余           -> invoke

# -*- coding: utf-8 -*-
"""
LangChain 1.x 的第一段代码：一次最短的问答
==========================================
这段代码做的事只有三件：
  1. 从环境变量里取密钥；
  2. 用统一入口 init_chat_model 拿到一个对话模型；
  3. invoke 一次，把模型返回的消息打印出来。

整段代码里没有任何一处在「生成文本」——生成是模型那一侧的事，
LangChain 只负责把请求组织好、把响应包成统一的消息对象。
"""

import os

from langchain.chat_models import init_chat_model

# 密钥一律走环境变量，不要写进源码，更不要提交进版本库。
# 终端里先执行：export OPENAI_API_KEY="sk-..."
if not os.environ.get("OPENAI_API_KEY"):
    raise SystemExit("请先设置环境变量 OPENAI_API_KEY")

# 如果走的是代理网关或第三方兼容端点，再额外设置 OPENAI_BASE_URL。
# init_chat_model 会把这两个环境变量透传给底层的 langchain-openai。
BASE_URL = os.environ.get("OPENAI_BASE_URL")

# "openai:gpt-4o-mini" 是 "提供方:模型名" 的写法。
# 换成 "ollama:qwen3:8b" 就切到本地模型，下面的代码一行都不用改。
model = init_chat_model(
    "openai:gpt-4o-mini",
    temperature=0.7,     # 随机性，0 最稳定，越大越发散
    max_tokens=512,      # 限制这次最多生成多少 token，防止跑飞
)

# invoke 接受字符串，也接受消息列表；这里给最简单的字符串。
response = model.invoke("用一句话解释什么是大模型")

# 返回的是一个 AIMessage 对象，而不是裸字符串。
print(type(response))          # <class 'langchain_core.messages.ai.AIMessage'>
print(response.content)        # 真正的文本内容

# 除了正文，响应里还带着用量信息，做成本核算时很有用。
print(response.usage_metadata)  # {'input_tokens': 12, 'output_tokens': 37, ...}

# 一个容易被忽略的事实：上面这次调用是无状态的。
# 再问一次「刚才我问了什么」，模型答不上来——它没有记忆，
# 所谓的多轮对话全靠你每轮把历史一起发过去。

# -*- coding: utf-8 -*-
"""
统一入口 init_chat_model：换模型只改一个字符串
================================================
LangChain 1.x 把「拿到一个对话模型」这件事收敛到了 langchain.chat_models
里的 init_chat_model。它的价值不在于少写几行，而在于：
把「用哪家模型」从代码问题降级成配置问题。
"""

import os

from langchain.chat_models import init_chat_model

# ---------------------------------------------------------------------------
# 写法一：一个字符串同时指定提供方与模型名，格式是 "提供方:模型名"
# ---------------------------------------------------------------------------
gpt = init_chat_model("openai:gpt-4o-mini", temperature=0)

# 换成本地 Ollama 拉起来的模型，只改这一个字符串。
# 前提是装了伙伴包 langchain-ollama，并且 ollama 服务已经在跑。
local = init_chat_model("ollama:qwen3:8b", temperature=0)

# ---------------------------------------------------------------------------
# 写法二：提供方与模型名分开传，适合模型名来自配置文件的场景
# ---------------------------------------------------------------------------
model_name = os.environ.get("CHAT_MODEL", "gpt-4o-mini")
provider = os.environ.get("CHAT_PROVIDER", "openai")

chat = init_chat_model(model_name, model_provider=provider, temperature=0.3)

# ---------------------------------------------------------------------------
# 写法三：运行时才决定用哪个模型
# configurable_fields 声明哪些字段允许在调用时改写，
# 之后用 with_config 就能在同一份代码里切换模型，不用重新构造对象。
# ---------------------------------------------------------------------------
switchable = init_chat_model(
    temperature=0,
    configurable_fields=("model", "model_provider"),
)

answer_a = switchable.with_config(
    configurable={"model": "gpt-4o-mini", "model_provider": "openai"}
).invoke("一句话介绍你自己")

answer_b = switchable.with_config(
    configurable={"model": "qwen3:8b", "model_provider": "ollama"}
).invoke("一句话介绍你自己")

print("云端模型：", answer_a.content)
print("本地模型：", answer_b.content)

# ---------------------------------------------------------------------------
# 为什么能统一？因为 init_chat_model 返回的对象都实现了同一套 Runnable 协议：
# invoke / batch / stream 以及对应的异步方法。
# 上层代码只依赖这套协议，不依赖某一家的 SDK 细节。
# ---------------------------------------------------------------------------
for chunk in gpt.stream("数三个数"):
    print(chunk.content, end="", flush=True)
print()

# 直接构造伙伴包里的类也完全可以，只是把提供方写死在了代码里：
#   from langchain_openai import ChatOpenAI
#   gpt = ChatOpenAI(model="gpt-4o-mini", temperature=0)
#   from langchain_ollama import ChatOllama
#   local = ChatOllama(model="qwen3:8b")
# 需要用到某家独有的参数时，直接构造更直白；其余情况优先用统一入口。

# -*- coding: utf-8 -*-
"""
PromptTemplate：把写死的提示词变成可填空的模板
================================================
提示词一旦写死，应用就只能回答一个问题。
模板的作用是把变化的部分挖成占位符，运行时再填进去。
"""

from langchain_core.prompts import PromptTemplate

# ---------------------------------------------------------------------------
# 两种实例化方式，效果完全一样
# ---------------------------------------------------------------------------
# 方式一：构造方法，显式声明变量名，适合模板来自配置、需要校验的场景
template_a = PromptTemplate(
    template="请简要描述{topic}的应用。",
    input_variables=["topic"],
)

# 方式二：from_template，自动从字符串里扫出 {} 占位符，日常写法
template_b = PromptTemplate.from_template("请简要描述{topic}的应用。")

print(template_a.input_variables)   # ['topic']
print(template_b.input_variables)   # ['topic']

# ---------------------------------------------------------------------------
# 填值：format 与 invoke 的区别
# ---------------------------------------------------------------------------
# format 返回字符串，人看着直观，调试时常用。
text = template_b.format(topic="机器学习")
print(type(text), text)             # <class 'str'> 请简要描述机器学习的应用。

# invoke 返回 PromptValue 对象，它能再转成字符串或消息列表，
# 所以能直接接到模型后面组成一条流水线。
value = template_b.invoke({"topic": "机器学习"})
print(type(value))                  # StringPromptValue
print(value.to_string())

# 记一个原则：自己看用 format，往下游传用 invoke。

# ---------------------------------------------------------------------------
# 多变量模板
# ---------------------------------------------------------------------------
review = PromptTemplate.from_template(
    "请评价{product}的优缺点，重点说{aspect1}和{aspect2}。"
)
print(review.format(product="笔记本电脑", aspect1="续航", aspect2="便携性"))

# 漏填任何一个变量都会直接报错，而不是把 {aspect2} 原样发给模型——
# 这正是模板相比字符串拼接的价值：错误在本地暴露，不会浪费一次调用。

# ---------------------------------------------------------------------------
# 部分变量：先固定一半，剩下的运行时再填
# ---------------------------------------------------------------------------
# 适合「角色、语气、输出格式」这类整轮不变的内容。
full = PromptTemplate.from_template(
    "你是一个{role}，请用{style}的风格回答：\n问题：{question}\n答案："
)

fixed = full.partial(role="资深厨师", style="专业但幽默")
print(fixed.format(question="如何煎牛排？"))     # 只需要再传 question

# 也可以在构造时就声明：
fixed2 = PromptTemplate(
    template="请评价{product}的优缺点，重点说{aspect1}和{aspect2}。",
    input_variables=["product"],
    partial_variables={"aspect1": "电池", "aspect2": "屏幕"},
)
print(fixed2.format(product="笔记本电脑"))

# ---------------------------------------------------------------------------
# 从文件加载模板
# ---------------------------------------------------------------------------
# 模板长到几十行时写在代码里很难维护，可以放进 yaml 或 json 单独管理，
# 好处是能单独做版本控制、能让非开发同事参与修改。
#   from langchain_core.prompts import load_prompt
#   prompt = load_prompt("asset/prompt.yaml", encoding="utf-8")
#   print(prompt.format(name="年轻人", what="滑稽"))

# -*- coding: utf-8 -*-
"""
ChatPromptTemplate：给多角色对话用的模板
=========================================
PromptTemplate 产出的是一整段字符串，ChatPromptTemplate 产出的是
一串带角色的消息。对话模型用后者，日常开发用得最多的也是后者。
"""

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate

# ---------------------------------------------------------------------------
# 两种实例化方式
# ---------------------------------------------------------------------------
# 列表里每个元素是 (角色, 内容) 的元组，角色写 "system" / "human" / "ai"。
chat_template = ChatPromptTemplate([
    ("system", "你是一个AI开发工程师，你的名字是{name}。"),
    ("human", "你能开发哪些AI应用?"),
    ("ai", "我能开发聊天机器人、图像识别、自然语言处理等应用。"),
    ("human", "{user_input}"),
])

# from_messages 的效果完全一样，只是写法更常见：
chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个AI开发工程师，你的名字是{name}。"),
    ("human", "你能开发哪些AI应用?"),
    ("ai", "我能开发聊天机器人、图像识别、自然语言处理等应用。"),
    ("human", "{user_input}"),
])

# 注意中间那两条 human / ai：它们不是真实发生过的对话，
# 而是写死在模板里的样例，用来给模型示范该怎么答。

# ---------------------------------------------------------------------------
# 三种填值方式，返回类型各不相同
# ---------------------------------------------------------------------------
# format：返回一整段字符串，各角色被拼成 "System: ... Human: ..." 的形式。
#         只适合肉眼检查，不要拿去喂对话模型。
as_text = chat_template.format(name="小智", user_input="你能帮我做什么?")
print(type(as_text))            # <class 'str'>
print(as_text)

# format_messages：返回消息对象列表，可以直接交给模型。
as_messages = chat_template.format_messages(name="小智", user_input="你能帮我做什么?")
print(type(as_messages), len(as_messages))    # <class 'list'> 4
print(as_messages[0])                          # SystemMessage(...)

# invoke：返回 ChatPromptValue，能再转字符串或消息列表，
#         也是能接进流水线的那一种。
as_value = chat_template.invoke({"name": "小智", "user_input": "你能帮我做什么?"})
print(type(as_value))                          # ChatPromptValue
print(len(as_value.messages))                  # 4

# 日常取舍：要拿去调模型就用 format_messages 或 invoke，
#           要打印出来看就用 format。

# ---------------------------------------------------------------------------
# 接上模型
# ---------------------------------------------------------------------------
model = init_chat_model("openai:gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{product}的客服助手，你的名字叫{name}，回答不超过两句话"),
    ("human", "{query}"),
])

messages = prompt.format_messages(product="一家咖啡店", name="小豆", query="你是谁")
print(model.invoke(messages).content)

# 这两步——格式化、调用——就是模型输入输出的前两段。
# 第三段解析在 output_parser_str_json.py 里；
# 至于怎么把三段接成一条流水线，是下一讲的事。

# ---------------------------------------------------------------------------
# 一个必须绕开的坑：花括号
# ---------------------------------------------------------------------------
# 模板里的 {} 一律被当成占位符。想在提示词里写出真正的花括号，要写两个。
literal = ChatPromptTemplate.from_messages([
    ("system", "请严格按这个结构返回：{{\"姓名\": \"\", \"年龄\": 0}}"),
    ("human", "{query}"),
])
print(literal.format_messages(query="介绍一下李雷"))

# -*- coding: utf-8 -*-
"""
MessagesPlaceholder：在模板中间留一个放消息列表的坑
=====================================================
模板里的普通占位符填的是一段文字，数量固定。
但对话历史的条数是运行时才知道的——可能 0 条，也可能 20 条。
MessagesPlaceholder 就是为这种「填进来的是一串消息」而准备的。
"""

from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# ---------------------------------------------------------------------------
# 基本用法
# ---------------------------------------------------------------------------
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位乐于助人的助手"),
    MessagesPlaceholder("history"),     # 这里会被展开成任意多条消息
    ("human", "{question}"),
])

# 传空历史：结果是 2 条消息（系统 + 当前提问）
print(len(prompt.format_messages(history=[], question="你好")))

# 传两条历史：结果是 4 条
history = [
    HumanMessage(content="1+2*3 等于几?"),
    AIMessage(content="等于 7"),
]
messages = prompt.format_messages(history=history, question="我刚才问的是什么?")
for m in messages:
    print(type(m).__name__, "|", m.content)

# 输出顺序是：系统消息 → 历史两条 → 当前提问。
# 位置由 MessagesPlaceholder 在模板里的位置决定，不会乱序。

# ---------------------------------------------------------------------------
# 接上模型：这就是「多轮对话」最朴素的实现
# ---------------------------------------------------------------------------
model = init_chat_model("openai:gpt-4o-mini")

chat_history = []          # 自己维护的一个列表

def ask(question):
    """问一轮，并把这一轮的问答塞回历史。"""
    msgs = prompt.format_messages(history=chat_history, question=question)
    answer = model.invoke(msgs)
    chat_history.append(HumanMessage(content=question))
    chat_history.append(answer)
    return answer.content

print(ask("我叫李雷，今年 18 岁"))
print(ask("我多大了?"))     # 模型答得上来，因为上一轮被重新发过去了

# 请注意这里发生了什么：模型没有任何记忆，
# 它能答出年龄，纯粹是因为 chat_history 里的两条消息又被完整发了一遍。
# 历史越长，每轮请求越贵——怎么裁剪、怎么摘要、怎么按会话隔离，
# 是后面「会话记忆」那一讲的正题。

# ---------------------------------------------------------------------------
# 可选占位：允许这一项不传
# ---------------------------------------------------------------------------
optional = ChatPromptTemplate.from_messages([
    ("system", "你是一位乐于助人的助手"),
    MessagesPlaceholder("history", optional=True),
    ("human", "{question}"),
])
print(len(optional.format_messages(question="你好")))   # 不传 history 也不报错

# -*- coding: utf-8 -*-
"""
少样本提示：不给规则，只给例子
================================
直接提要求叫零样本（zero-shot），先给几个示范再提要求叫少样本（few-shot）。
当你发现「怎么说它都不按格式来」的时候，少样本往往比把规则写得更长更管用。
"""

from langchain.chat_models import init_chat_model
from langchain_core.prompts import (ChatPromptTemplate,
                                    FewShotChatMessagePromptTemplate,
                                    FewShotPromptTemplate, PromptTemplate)

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

# ---------------------------------------------------------------------------
# 先看零样本会怎么翻车
# ---------------------------------------------------------------------------
print(model.invoke("北京天气怎么样").content)
# 模型会长篇大论地解释它查不到实时天气。
# 但假设我们只想从这句话里抽出城市名，上面这个回答毫无用处。

# ---------------------------------------------------------------------------
# 一、FewShotPromptTemplate：拼一整段字符串提示词
# ---------------------------------------------------------------------------
# 第一步：写清楚「单个示例长什么样」
example_prompt = PromptTemplate.from_template("输入：{input}\n输出：{output}")

# 第二步：准备示例集合，每个示例是一个字典，键与上面的变量名一一对应
examples = [
    {"input": "北京天气怎么样", "output": "北京市"},
    {"input": "南京下雨吗", "output": "南京市"},
    {"input": "武汉热吗", "output": "武汉市"},
]

# 第三步：组装。prefix 放在所有示例前面，suffix 放在最后，留出真正的提问
few_shot = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="从下面这句话里提取城市名，只回答城市名。",
    suffix="输入：{input}\n输出：",
    input_variables=["input"],
)

prompt_text = few_shot.format(input="长沙多少度")
print(prompt_text)
# 渲染出来是：说明 + 三组示范 + 待回答的那一条，末尾停在「输出：」等着模型接话。

print(model.invoke(prompt_text).content)     # 长沙市

# 三个例子就把输出格式钉死了，而且一个字的规则都没写。

# ---------------------------------------------------------------------------
# 二、FewShotChatMessagePromptTemplate：把示例摊成多轮消息
# ---------------------------------------------------------------------------
# 对话模型更吃这一套：示例被展开成一问一答的消息对，像是真的聊过。
example_messages = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot_chat = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=example_messages,
)

final = ChatPromptTemplate.from_messages([
    ("system", "你负责从用户的话里提取城市名，只回答城市名。"),
    few_shot_chat,                  # 三组示范会在这里展开成 6 条消息
    ("human", "{input}"),
])

msgs = final.format_messages(input="长沙多少度")
print(len(msgs))                    # 1 条系统 + 6 条示范 + 1 条提问 = 8
print(model.invoke(msgs).content)

# ---------------------------------------------------------------------------
# 用几个例子合适
# ---------------------------------------------------------------------------
# 2 到 5 个通常就够。例子太多有两个副作用：
#   一是每轮都要把它们重新发一遍，token 成本线性上涨；
#   二是例子覆盖面不均时，模型会被带偏到某一类上去。
# 例子的质量比数量重要：优先挑边界情况，别全是最典型的那种。

# -*- coding: utf-8 -*-
"""
输出解析器：把模型的话变成程序能用的数据
==========================================
模型返回的是消息对象，正文是一段自然语言。
但下游代码要的往往是一个字符串、一个字典、一个列表。
解析器干的就是这一步转换，顺带还能反过来约束模型该怎么输出。
"""

from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import (CommaSeparatedListOutputParser,
                                           JsonOutputParser, StrOutputParser)
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

# ---------------------------------------------------------------------------
# 一、StrOutputParser：只把正文抠出来
# ---------------------------------------------------------------------------
raw = model.invoke("把这句话翻译成中文：It's a nice day today")
print(type(raw))            # AIMessage，带一堆用量与元信息

parser = StrOutputParser()
text = parser.invoke(raw)
print(type(text), text)     # <class 'str'> 今天天气不错。

# 它做的事等价于 raw.content，价值在于「它也是一个 Runnable」——
# 能作为流水线的最后一节挂上去，让整条链的输出类型稳定是字符串。

# ---------------------------------------------------------------------------
# 二、JsonOutputParser：要结构化数据
# ---------------------------------------------------------------------------
json_parser = JsonOutputParser()

# 关键在 get_format_instructions()：它生成一段说明书，告诉模型该怎么排版输出。
# 把这段说明塞进提示词，模型才会乖乖返回 JSON。
print(json_parser.get_format_instructions())

prompt = PromptTemplate(
    template="回答用户的问题。\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": json_parser.get_format_instructions()},
)

resp = model.invoke(prompt.format(query="讲一个笑话，问题用 q 表示，答案用 a 表示"))
data = json_parser.invoke(resp)
print(type(data), data)     # <class 'dict'> {'q': '...', 'a': '...'}
print(data["a"])            # 可以直接按键取值

# 解析器还顺手擦了屁股：模型常常把 JSON 包在代码块标记里，
# 直接 json.loads 会报错，JsonOutputParser 会先把这层壳剥掉。

# ---------------------------------------------------------------------------
# 三、CommaSeparatedListOutputParser：要一个列表
# ---------------------------------------------------------------------------
list_parser = CommaSeparatedListOutputParser()

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位心理学助手"),
    ("human", "{request}\n{format_instructions}"),
])

msgs = chat_prompt.format_messages(
    request="给我 5 种情绪",
    format_instructions=list_parser.get_format_instructions(),
)
moods = list_parser.invoke(model.invoke(msgs))
print(type(moods), moods)   # <class 'list'> ['快乐', '忧伤', '愤怒', '兴奋', '宁静']

# ---------------------------------------------------------------------------
# 还有哪些解析器
# ---------------------------------------------------------------------------
# XMLOutputParser              要 XML 结构
# DatetimeOutputParser         把回答解析成 datetime 对象
# EnumOutputParser             限定在若干枚举值里选一个
# PydanticOutputParser         解析成 Pydantic 模型实例，带字段校验
# OutputFixingParser           解析失败时，再调一次模型把格式修好
#
# 共同点是它们都实现了 Runnable 协议，所以用法一致，
# 都可以挂在模型后面当流水线的最后一节。

# -*- coding: utf-8 -*-
"""
结构化输出：让模型的回答带字段校验
====================================
JsonOutputParser 只保证「是一个 JSON」，不保证字段齐、类型对。
真要把模型输出接进业务逻辑，得再往前走一步。
"""

from typing import List

from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

model = init_chat_model("openai:gpt-4o-mini", temperature=0)


# ---------------------------------------------------------------------------
# 先把「我要什么」用一个类声明清楚
# ---------------------------------------------------------------------------
class Book(BaseModel):
    """一本书的基本信息。"""

    title: str = Field(description="书名")
    author: str = Field(description="作者姓名")
    year: int = Field(description="出版年份，四位数字")
    tags: List[str] = Field(description="三个主题标签")


# description 不是写给同事看的注释，模型会读它来决定每个字段填什么，
# 写含糊了，字段就容易填错或者空着。

# ---------------------------------------------------------------------------
# 方式一：PydanticOutputParser
# ---------------------------------------------------------------------------
parser = PydanticOutputParser(pydantic_object=Book)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个图书信息助手。\n{format_instructions}"),
    ("human", "介绍一下《三体》"),
]).partial(format_instructions=parser.get_format_instructions())

book = parser.invoke(model.invoke(prompt.format_messages()))
print(type(book))          # <class '__main__.Book'>
print(book.title, book.year, book.tags)

# 拿到的是一个真正的对象，可以点出字段，类型也已经被校验过：
# year 如果模型返回了 "一九九一年" 这种字符串，这一步就会直接报错，
# 而不是等到下游做减法时才炸。

# ---------------------------------------------------------------------------
# 方式二：with_structured_output，把约束下沉到模型层
# ---------------------------------------------------------------------------
structured = model.with_structured_output(Book)
book2 = structured.invoke("介绍一下《活着》")
print(book2.author, book2.tags)

# 两者的差别值得记一下：
#   PydanticOutputParser  靠提示词说服模型排版，任何模型都能用，
#                         但模型不听话时就得靠 OutputFixingParser 兜底。
#   with_structured_output 由提供方在协议层保证结构，可靠得多，
#                         前提是这家模型支持结构化输出或工具调用。
# 能用后者就用后者；接的是本地小模型或老接口，再退回前者。

# ---------------------------------------------------------------------------
# 解析失败怎么办
# ---------------------------------------------------------------------------
# 不要直接把 parser.invoke 裸放在生产路径上，至少包一层：
try:
    result = parser.invoke(model.invoke(prompt.format_messages()))
except Exception as exc:
    print("解析失败，走降级路径：", exc)
    result = None

# 需要自动修复时用 OutputFixingParser，它会把出错的原文连同报错一起
# 再发给模型，让模型自己改对——代价是多一次调用：
#   from langchain_classic.output_parsers import OutputFixingParser
#   safe = OutputFixingParser.from_llm(parser=parser, llm=model)

# -*- coding: utf-8 -*-
"""
完整案例：把一条商品评论拆成结构化字段
========================================
场景：电商后台每天收上来几万条评论，运营要的不是原文，
而是「情感倾向、涉及哪些方面、要不要人工跟进」这三样能进报表的字段。

这条路径把模型输入输出的三段走了一遍：
  格式化（把评论塞进模板） → 预测（调模型） → 解析（转成对象）
"""

import os
from typing import List, Literal

from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

if not os.environ.get("OPENAI_API_KEY"):
    raise SystemExit("请先设置环境变量 OPENAI_API_KEY")


# ---------------------------------------------------------------------------
# 第一步：把「要什么」声明清楚
# ---------------------------------------------------------------------------
class ReviewInsight(BaseModel):
    """一条评论的结构化结论。"""

    sentiment: Literal["正面", "负面", "中性"] = Field(description="整体情感倾向")
    aspects: List[str] = Field(description="评论涉及的方面，如 物流、包装、续航、客服")
    complaint: str = Field(description="用户抱怨的核心问题，没有抱怨则填写 无")
    need_followup: bool = Field(description="是否需要人工跟进，涉及退换货或投诉时为真")


parser = PydanticOutputParser(pydantic_object=ReviewInsight)

# ---------------------------------------------------------------------------
# 第二步：格式化。系统消息定规矩，少量示范钉死口径
# ---------------------------------------------------------------------------
prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是电商评论分析助手。只依据评论原文判断，不要脑补没写的内容。\n"
     "{format_instructions}"),
    ("human", "评论：手机很好用，就是快递等了五天，包装也压扁了。"),
    ("ai",
     '{{"sentiment": "中性", "aspects": ["物流", "包装"], '
     '"complaint": "快递慢且包装破损", "need_followup": false}}'),
    ("human", "评论：{review}"),
]).partial(format_instructions=parser.get_format_instructions())

# 注意示范那条 ai 消息里的双花括号：模板会把 {} 当占位符，
# 想输出真正的花括号必须写两个。

# ---------------------------------------------------------------------------
# 第三步：预测 + 解析
# ---------------------------------------------------------------------------
model = init_chat_model("openai:gpt-4o-mini", temperature=0)


def analyze(review: str) -> ReviewInsight:
    """把一条评论转成结构化结论。"""
    messages = prompt.format_messages(review=review)
    response = model.invoke(messages)          # 预测：拿回 AIMessage
    return parser.invoke(response)             # 解析：转成 ReviewInsight 对象


one = analyze("用了两周电池掉得飞快，申请换货客服一直不回，太失望了。")
print(one.sentiment)        # 负面
print(one.aspects)          # ['续航', '客服']
print(one.need_followup)    # True

# ---------------------------------------------------------------------------
# 批量处理：一次提交多条，并发跑
# ---------------------------------------------------------------------------
reviews = [
    "屏幕清晰，手感也好，价格算是同级里厚道的。",
    "发货倒是快，可是收到就是划痕机，要求退货。",
    "还行吧，没什么特别的感觉。",
]

batch_messages = [prompt.format_messages(review=r) for r in reviews]
responses = model.batch(batch_messages, config={"max_concurrency": 3})

for text, resp in zip(reviews, responses):
    try:
        item = parser.invoke(resp)
    except Exception as exc:                   # 模型偶尔会输出不合规的结构
        print("解析失败，转人工：", text[:12], exc)
        continue
    flag = "需跟进" if item.need_followup else "无需跟进"
    print(f"{item.sentiment} | {'、'.join(item.aspects)} | {flag}")

# ---------------------------------------------------------------------------
# 几个让这段代码能上生产的细节
# ---------------------------------------------------------------------------
# 1. temperature 必须压到 0。这是抽取任务，同一条评论每次都该得到同样的结论。
# 2. 解析一定要包 try：模型不是接口，它没有义务保证结构合法。
# 3. 失败的那条要能落到人工队列，别静默丢掉。
# 4. 字段的 description 是给模型读的，「涉及的方面」写得越具体，分类越稳。

# -*- coding: utf-8 -*-
"""
接本地私有模型：Ollama
========================
数据不能出内网、想省掉按 token 计费、需要离线演示——这三种情况下
都会把模型换成本地部署。对上层代码来说，换的只是一个名字。

准备工作（一次性）：
  1. 装 Ollama：Linux 执行 curl -fsSL https://ollama.com/install.sh | sh
  2. 拉模型并跑起来：ollama run qwen3:8b
  3. 装伙伴包：pip install langchain-ollama
"""

from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama, OllamaEmbeddings

# ---------------------------------------------------------------------------
# 写法一：还是统一入口，只换提供方
# ---------------------------------------------------------------------------
model = init_chat_model("ollama:qwen3:8b", temperature=0)
print(model.invoke("你好，请介绍一下你自己").content)

# ---------------------------------------------------------------------------
# 写法二：直接构造伙伴包里的类，需要用 Ollama 特有参数时用它
# ---------------------------------------------------------------------------
local = ChatOllama(
    model="qwen3:8b",
    temperature=0,
    base_url="http://127.0.0.1:11434",   # 服务不在本机默认端口时才需要指定
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译助手，把{input_language}翻译成{output_language}，只输出译文"),
    ("human", "{text}"),
])

messages = prompt.format_messages(input_language="中文", output_language="英语", text="我爱编程")
print(local.invoke(messages).content)

# 注意：整个提示词模板、消息结构、解析器一个字都没改。
# 这正是统一接口的意义——后厨换了灶台，前面的工序不用重排。

# ---------------------------------------------------------------------------
# 本地嵌入模型
# ---------------------------------------------------------------------------
emb = OllamaEmbeddings(model="bge-m3")
vector = emb.embed_query("这是第一个测试文档")
print(len(vector))          # 向量维度，由模型决定

# ---------------------------------------------------------------------------
# 三个实际会踩到的差异
# ---------------------------------------------------------------------------
# 1. 带思考过程的模型（例如 deepseek-r1）会先吐一段推理再给答案，
#    直接拿 content 会把推理也带上。1.x 里可以按内容块类型挑出正文：
resp = local.invoke("1 加 1 等于几")
for block in resp.content_blocks:
    if block["type"] == "text":
        print(block["text"])

# 2. 本地小模型对格式要求的遵从度明显低于云端大模型，
#    JSON 解析失败的概率更高，少样本示例和解析失败兜底更有必要。
#
# 3. 首次调用要把模型权重读进显存，会慢十几秒，别误判成程序卡死。

# ---------------------------------------------------------------------------
# 旧写法提示
# ---------------------------------------------------------------------------
# 早期资料里的 from langchain_community.llms import Ollama 与
# from langchain_community.chat_models import ChatOllama 仍能跑，
# 但官方伙伴包 langchain-ollama 才是现在的推荐入口，功能跟进也更快。

chain_style_output = StrOutputParser().invoke(local.invoke("说一句话"))
print(chain_style_output)

# -*- coding: utf-8 -*-
"""
骨架模板：模型输入输出三段式，复制改 TODO 即可
===============================================
把本讲所有例子的共性抽出来就是这一份：
  格式化 -> 预测 -> 解析，外加一层失败兜底。

改动点只有五处，都标了 TODO；其余代码不用动。
"""

import os
from typing import List

from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

# ---------------------------------------------------------------------------
# TODO 1：换模型。
#   云端： "openai:gpt-4o-mini"   本地： "ollama:qwen3:8b"
#   抽取、分类这类任务 temperature 一律 0；写文案才调高。
# ---------------------------------------------------------------------------
MODEL_NAME = os.environ.get("CHAT_MODEL", "openai:gpt-4o-mini")
model = init_chat_model(MODEL_NAME, temperature=0)


# ---------------------------------------------------------------------------
# TODO 2：声明你要的结构。
#   每个字段的 description 是写给模型看的，写清楚含义、格式和取值范围。
#   只要一段纯文本的话，把这个类连同下面的 parser 一起删掉，
#   换成 parser = StrOutputParser() 即可。
# ---------------------------------------------------------------------------
class Result(BaseModel):
    """这里写清楚这个结构整体代表什么。"""

    field_a: str = Field(description="字段含义，写明格式与示例")
    field_b: List[str] = Field(description="列表字段，说明元素是什么")
    field_c: bool = Field(description="布尔字段，说明什么情况下为真")


parser = PydanticOutputParser(pydantic_object=Result)
text_parser = StrOutputParser()      # 只要纯文本时改用这个

# ---------------------------------------------------------------------------
# TODO 3：写提示词。
#   system 定身份、定规矩、放格式说明；
#   中间可以插入若干组 human / ai 示范（少样本），不需要就删掉；
#   最后一条 human 留占位符，运行时填真实输入。
# ---------------------------------------------------------------------------
prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是一个TODO角色。只依据输入内容作答，不要编造没有的信息。\n"
     "{format_instructions}"),
    ("human", "TODO：一条示范输入"),
    ("ai", '{{"field_a": "示范值", "field_b": ["示范"], "field_c": false}}'),
    ("human", "{user_input}"),
]).partial(format_instructions=parser.get_format_instructions())


# ---------------------------------------------------------------------------
# 三段式主流程：格式化 -> 预测 -> 解析
# ---------------------------------------------------------------------------
def run(user_input: str):
    """处理单条输入，解析失败时返回 None 由调用方决定怎么降级。"""
    messages = prompt.format_messages(user_input=user_input)   # 一、格式化
    response = model.invoke(messages)                          # 二、预测
    try:
        return parser.invoke(response)                         # 三、解析
    except Exception as exc:
        print("解析失败：", exc)
        print("模型原文：", response.content[:200])
        return None


def run_many(inputs: List[str]):
    """批量处理。并发数按提供方的频率限制调，别一次打太多。"""
    batch = [prompt.format_messages(user_input=x) for x in inputs]
    responses = model.batch(batch, config={"max_concurrency": 5})
    out = []
    for raw, response in zip(inputs, responses):
        try:
            out.append(parser.invoke(response))
        except Exception:
            out.append(None)                # 失败的位置留空，方便回查是哪一条
    return out


if __name__ == "__main__":
    # TODO 4：换成你的真实输入
    print(run("TODO：这里换成真实的用户输入"))

    # TODO 5：需要流式展示时，把上面的 invoke 换成 stream。
    #   注意：流式与结构化解析不能同时要——没收全的 JSON 解析不了。
    #   界面上要边打字边显示，就只能用纯文本解析器。
    for chunk in model.stream("说一句话"):
        print(chunk.content, end="", flush=True)
    print()

# -*- coding: utf-8 -*-
"""
旧写法与现写法：读老资料时怎么换算
====================================
网上大量示例停留在 0.3 时代。它们的思路依然成立，但导入路径和类名
已经变了。这份文件把你最容易撞上的几处放在一起，方便照着改。
"""

# ---------------------------------------------------------------------------
# 一、导入路径：提示词与解析器都在 langchain-core 里
# ---------------------------------------------------------------------------
# 旧：from langchain.prompts import PromptTemplate, ChatPromptTemplate
# 旧：from langchain.prompts.few_shot import FewShotPromptTemplate
# 现：
from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_core.prompts import (ChatPromptTemplate, FewShotPromptTemplate,
                                    MessagesPlaceholder, PromptTemplate)

# 判断标准：凡是「不依赖任何具体模型提供方」的基础抽象——
# 运行协议、消息、提示词、工具定义、输出解析器——都归 langchain-core。

# ---------------------------------------------------------------------------
# 二、获取模型：主包给了统一入口
# ---------------------------------------------------------------------------
# 旧：from langchain_community.llms import Ollama
#     model = Ollama(model="qwen3:8b")
# 旧：from langchain_community.chat_models import ChatOllama
# 现：官方伙伴包，或者直接用统一入口
from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings

model = init_chat_model("ollama:qwen3:8b", temperature=0)
emb = init_embeddings("ollama:bge-m3")

# langchain-community 没有消失，它仍然是第三方集成的大仓库；
# 只是常用的那几家已经各自有了官方伙伴包，优先用伙伴包。

# ---------------------------------------------------------------------------
# 三、legacy 链：搬到了 langchain-classic
# ---------------------------------------------------------------------------
# 旧：from langchain.chains import LLMChain
#     chain = LLMChain(llm=model, prompt=prompt)
#     chain.run(topic="春天")
#
# 这些类没有从世界上消失，它们搬到了 langchain-classic：
#     from langchain_classic.chains import LLMChain
#
# 但它们不再是推荐写法。现在组合组件靠的是管道符——
# 怎么串、串出来是什么、为什么这么设计，是下一讲的正题。
prompt = PromptTemplate.from_template("写一句关于{topic}的话")
chain = prompt | model | StrOutputParser()
print(chain.invoke({"topic": "春天"}))

# 同样搬进 langchain-classic 的还有：
#   ConversationChain、ConversationBufferMemory 等会话记忆相关的类
#   MultiQueryRetriever 等检索器
#   indexing API、hub 模块、缓存嵌入辅助类
#   from langchain import hub  ->  from langchain_classic import hub

# ---------------------------------------------------------------------------
# 四、主包 1.x 还留着什么
# ---------------------------------------------------------------------------
# langchain.agents      create_agent、AgentState
# langchain.messages    消息类型、内容块、trim_messages
# langchain.tools       tool 装饰器、BaseTool
# langchain.chat_models init_chat_model、BaseChatModel
# langchain.embeddings  init_embeddings、Embeddings
#
# 换句话说：主包现在只管「搭认知架构」这一件事，其余都下沉或外移了。

# ---------------------------------------------------------------------------
# 五、还有一处小陷阱
# ---------------------------------------------------------------------------
# 老代码里能见到 model(messages) 这种把模型对象当函数调的写法，
# 它走的是内部的兼容入口，最终还是转到 invoke。新代码一律写 invoke。
resp = model.invoke("你好")
print(resp.content)

# 版本查验永远比猜测靠谱：
#   import langchain, langchain_core
#   print(langchain.__version__, langchain_core.__version__)