LangChain入门（一）

最近在学 LangChain，踩了不少坑，也有一些"原来这么简单"的时刻。把学到的东西整理一下，从最基础的模型调用，到提示词模板、输出解析器、链式调用，最后把应用部署成 Web 服务。

1. 模型包装器：统一调用各家大模型

LangChain 做的第一件事就是把各家大模型的 API 包了一层。不管你用通义千问、DeepSeek 还是腾讯混元，调用方式都一样。

最简方式：直接用社区模型

from langchain_community.chat_models import ChatTongyi

model = ChatTongyi()
response = model.invoke("你是谁？")
print(response.content)

前提是你配好了环境变量 DASHSCOPE_API_KEY。跑起来没问题，但灵活度不够。

通用方式：ChatOpenAI 兼容各平台

大部分国内模型都兼容 OpenAI 的接口规范，所以用 ChatOpenAI 一把梭：

import os
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-max-latest",
    temperature=0.7
)

result = model.invoke("用一句话解释什么是LangChain")
print(result.content)

换 DeepSeek？改三个参数就行：

model = ChatOpenAI(
    api_key=os.getenv("Deepseek_Key"),
    base_url="https://api.deepseek.com/v1",
    model="deepseek-chat",
    temperature=0.7
)

我自己封装了一个工具函数，省得每次写一堆：

def get_lc_model_client(api_key, base_url, model, temperature=0.7):
    return ChatOpenAI(
        api_key=api_key,
        base_url=base_url,
        model=model,
        temperature=temperature
    )

2. 提示词模板：别再手动拼字符串了

硬编码提示词写死在代码里，改起来痛苦。LangChain 提供了模板机制，用占位符动态填充内容。

字符串模板

最简单的形式，一个模板字符串加几个变量：

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate(
    template="你是一个翻译助手，请将以下内容翻译成{language}：{text}"
)

# 填入实际内容
actual_prompt = prompt.format(language="中文", text="I am a programmer")
print(actual_prompt)
# 输出：你是一个翻译助手，请将以下内容翻译成中文：I am a programmer

对话模板（多角色）

真实场景里，大模型对话有角色区分——system 设定身份，human 是用户输入。用 ChatPromptTemplate：

from langchain_core.prompts import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate
)

prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template(
        "你是一个翻译助手，请将以下内容翻译成{language}"
    ),
    HumanMessagePromptTemplate.from_template("{text}")
])

actual_prompt = prompt.format(language="中文", text="I am a programmer")

也可以用简写形式，效果一样：

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译助手，请将以下内容翻译成{language}"),
    ("human", "{text}")
])

少样本提示（Few-shot）

有时候你需要给模型"举几个例子"，让它学会你想要的输出格式：

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate

# 准备示例
examples = [
    {"sinput": "2+2", "soutput": "4", "sdescription": "加法运算"},
    {"sinput": "5-2", "soutput": "3", "sdescription": "减法运算"},
]

# 单个示例的格式模板
example_template = PromptTemplate(
    template="算式：{sinput} 值：{soutput} 类型：{sdescription}"
)

# 组装少样本模板
prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_template,
    prefix="你是一个数学专家，能够准确说出算式的类型，",
    suffix="现在给你算式：{input}，值：{output}，告诉我类型：",
    input_variables=["input", "output"]
)

result = model.invoke(prompt.format(input="2*5", output="10"))
print(result.content)  # 乘法运算

模型看了加法和减法的例子，就能推断出 2*5 是乘法。

Partial 变量：分步填充模板

有些参数你想提前固定，剩下的后面再填。比如做一个"学习助手"，先锁定学科，再接收用户的具体问题。用 prompt.partial() 就行：

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate(
    template="讲一个关于{date}的小故事：{text}",
    input_variables=["date", "text"]
)

# 先固定 date 参数
half_prompt = prompt.partial(date="2008-08-08")

# 后续只需要填 text
result = model.invoke(half_prompt.format(text="一个幸福的爱情故事"))
print(result.content)

3. 链（Chain）：用管道符串联整个流程

LangChain 里最优雅的设计是链式调用。用 | 把提示词模板、模型、解析器串起来，数据自动流转：

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 三个组件
prompt = PromptTemplate(
    template="你是一个翻译助手，请将以下内容翻译成{language}：{text}"
)
model = ChatOpenAI(
    api_key="your-api-key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-max-latest"
)
parser = StrOutputParser()

# 一行搞定
chain = prompt | model | parser

# 调用
result = chain.invoke({"language": "中文", "text": "I am a programmer"})
print(result)  # 我是一个程序员

数据流向：dict → prompt模板填充 → model推理 → parser提取文本。不用手动传中间结果，代码干净很多。

4. 输出解析器：把模型的回答变成结构化数据

大模型返回的是自然语言文本。如果你需要 JSON、列表这类结构化数据，就要用解析器。

StrOutputParser：提取纯文本

模型返回的是 AIMessage 对象，解析器帮你拿到 .content 字符串：

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
chain = prompt | model | parser
# 返回的直接是字符串，不是 AIMessage 对象

JsonOutputParser：拿到 dict

告诉模型"用 JSON 格式返回"，然后用 JSON 解析器接：

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的程序员"),
    ("user", "{input}")
])

parser = JsonOutputParser()
chain = prompt | model | parser

result = chain.invoke({
    "input": "langchain是什么？用question表示问题，用ans表示回答，返回JSON格式"
})
print(type(result))  # <class 'dict'>
print(result)        # {'question': 'langchain是什么？', 'ans': '...'}

注意：你得在提示词里明确说"返回 JSON 格式"，不然模型可能返回普通文本，解析器会报错。

CommaSeparatedListOutputParser：拿到列表

让模型用逗号分隔回答，解析器帮你拆成 Python 列表：

from langchain_core.output_parsers import CommaSeparatedListOutputParser

parser = CommaSeparatedListOutputParser()
chain = prompt | model | parser

result = chain.invoke({
    "input": "列出Python的三个主要版本，用逗号分隔"
})
print(result)  # ['Python 2', 'Python 3.x', 'Python 3.12']

DatetimeOutputParser：拿到日期对象

问模型时间相关的问题，直接拿到 Python 的 datetime 对象：

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import DatetimeOutputParser

output_parser = DatetimeOutputParser()

template = """
回答用户的问题：{question}

{format_instructions}
"""

prompt = PromptTemplate.from_template(
    template,
    partial_variables={
        "format_instructions": output_parser.get_format_instructions()
    },
)

chain = prompt | model | output_parser
result = chain.invoke({"question": "新中国是什么时候成立的？"})
print(result)  # 1949-10-01 00:00:00

这里 get_format_instructions() 会生成一段提示文本，告诉模型用什么格式回答日期。解析器再把字符串转成 datetime。

自定义解析器：内置的不够用就自己写

LangChain 内置的解析器覆盖了常见场景，但总有它们搞不定的时候。比如你想让模型返回特定格式的日期字符串，还要做校验，内置的 DatetimeOutputParser 不支持自定义格式。这时候继承 BaseOutputParser 自己写一个：

import datetime
from langchain_core.output_parsers import BaseOutputParser
from langchain_core.exceptions import OutputParserException


class DateStringParser(BaseOutputParser[str]):
    """自定义日期解析器，支持任意日期格式"""

    # 期望的日期格式，可以在实例化时指定
    target_format: str = "%Y-%m-%d"

    def parse(self, text: str) -> str:
        """把模型返回的文本解析成指定格式的日期字符串"""
        stripped_text = text.strip()
        try:
            dt_object = datetime.datetime.strptime(stripped_text, self.target_format)
            return dt_object.strftime(self.target_format)
        except ValueError as e:
            raise OutputParserException(
                f"日期格式不对。期望 '{self.target_format}'，收到: '{stripped_text}'. 错误: {e}"
            )

    def get_format_instructions(self) -> str:
        """告诉模型应该怎么输出"""
        example_date = datetime.datetime.now().strftime(self.target_format)
        return (
            f"请严格按照以下格式输出日期：'{self.target_format}'.\n"
            f"例如: {example_date}\n"
            f"只返回日期字符串，不要包含任何其他文本！"
        )

两个方法各司其职：get_format_instructions() 生成提示词片段，约束模型的输出格式；parse() 拿到模型返回的文本后做校验和转换。格式不对直接抛异常，不会让脏数据溜过去。

用起来和内置解析器一样，塞进链里就行：

from langchain_core.prompts import PromptTemplate

# 想要 月/日/年 时:分:秒 这种格式？改 target_format 就行
date_parser = DateStringParser(target_format="%m/%d/%Y %H:%M:%S")

template = """
回答用户的问题：{question}

{format_instructions}
"""

prompt = PromptTemplate.from_template(
    template,
    partial_variables={"format_instructions": date_parser.get_format_instructions()},
)

chain = prompt | model | date_parser

result = chain.invoke({"question": "北京夏季奥运会的开幕时间是？"})
print(result)  # 08/08/2008 20:00:00

写自定义解析器的套路就这些：继承 BaseOutputParser，实现 parse() 和 get_format_instructions()。前者负责"怎么解析"，后者负责"怎么告诉模型按格式输出"。想解析什么格式都行，XML、YAML、自定义 DSL，逻辑你自己定。

5. 部署：把 Chain 变成 Web 服务

写好的 chain 想给别人用？LangServe 能把它直接部署成 HTTP 接口。

服务端

from fastapi import FastAPI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate
from langchain_openai import ChatOpenAI
from langserve import add_routes

# 组装 chain
model = ChatOpenAI(
    api_key="your-api-key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-max-latest"
)

prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("请将以下的内容翻译成{language}"),
    ("human", "{text}")
])

chain = prompt | model | StrOutputParser()

# 部署
app = FastAPI(title="翻译服务", version="1.0")
add_routes(app, chain, path="/translate")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

启动后访问 http://localhost:8000/docs 能看到自动生成的 API 文档。

客户端调用

Python 客户端用 RemoteRunnable：

from langserve import RemoteRunnable

client = RemoteRunnable("http://localhost:8000/translate")
result = client.invoke({"language": "意大利文", "text": "我喜欢编程"})
print(result)

其他语言直接发 HTTP 请求也行：

curl -X POST http://localhost:8000/translate/invoke \
  -H "Content-Type: application/json" \
  -d '{"input": {"language": "意大利文", "text": "我喜欢编程"}}'

整体架构回顾

用户输入
  ↓
PromptTemplate（模板填充）
  ↓
ChatOpenAI（模型推理）
  ↓
OutputParser（结构化输出）
  ↓
LangServe（对外服务）

LangChain 的思路就是把 LLM 应用拆成可组合的模块。每个模块干一件事，用 | 串起来。改模型？换一行。改输出格式？换个解析器。加个前处理？插到链里。

调试：看看链里到底发生了什么

链式调用写起来简洁，但出了问题不好排查——中间结果看不见。LangChain 提供了全局调试开关，打开后会把每一步的输入输出都打印出来：

from langchain_core.globals import set_debug

set_debug(True)

加这两行就行，放在代码最前面。之后跑任何 chain，控制台会输出每个组件收到什么、返回什么。prompt 拼出来长什么样、模型原始返回是什么、解析器处理前后的变化，全都能看到。

调试完记得关掉，不然日志刷屏：

set_debug(False)

开发阶段建议常开，部署时关闭。

环境配置备忘

pip install langchain langchain-openai langchain-community langserve fastapi uvicorn

API Key 建议放环境变量里，别写代码里：

export DASHSCOPE_API_KEY="sk-xxx"
export Deepseek_Key="sk-xxx"

写到这里差不多了。LangChain 的 Model I/O 模块就是干这些事——包装模型、格式化提示词、解析输出、组成链、部署服务。后面还有 RAG、Agent、Memory 这些更有意思的东西，等学到了再写。