☰
LangChain入门与Model I/O实战:用TaoToken统一Key跑通PromptTemplate与LCEL(附完整代码)
2026/9/29 3:54:10 网站建设 项目流程

1. 从手写 API 到 LangChain:Model I/O 到底解决什么问题

如果你已经能自己用 requests 或 openai SDK 调通大模型接口,接下来大概率会遇到一个尴尬:每换一家模型,就要重写一遍调用代码;Prompt 越写越长,字符串拼接开始失控;模型返回一段带解释的 JSON,你还得写正则去抠。LangChain 的 Model I/O 就是冲着这些重复劳动来的,它把「提示词 → 模型 → 输出解析」这条链路抽象成三个可替换的组件,让你把精力放在业务逻辑上。

这篇面向的是有 Python 基础、想从零搭一套大模型应用开发框架的开发者。我会用 TaoToken 作为统一的 Key 和 API 通道,把 PromptTemplate、ChatModel、OutputParser 以及 LCEL 管道串起来,最后交付一份可复制的 config 骨架和 settings.json 片段,并做一次端到端调用验证。整套流程跑通后,你换模型只需要改配置里的一行,业务代码基本不动。

需要提前说明的是,LangChain 的版本迭代很快,本文基于 v1.x 的接口习惯编写。如果你本地是更早的版本,部分导入路径可能不同,建议先升级到较新的稳定版再跟着操作。下面所有代码都可以直接复制运行,唯一需要你替换的是自己的 API Key。

2. TaoToken 前置准备:统一 Key 与 API 通道

2.1 为什么用统一通道而不是直连各家

原生 SDK 的痛点在于「一家一套写法」。DeepSeek 用https://api.deepseek.com,通义用 dashscope 的兼容地址,智谱又是另一个域名。业务里同时对接三家,就要维护三套 client 初始化、三套错误处理、三套重试逻辑。TaoToken 提供的是 OpenAI 兼容的统一入口,你只需要记住一个 base_url 和一个 Key,模型名通过参数切换,LangChain 侧完全感知不到底层差异。

对 LangChain 来说,这一点尤其重要:ChatOpenAI这个类本身就支持自定义base_url,所以只要通道是 OpenAI 兼容的,就能直接接进来,不需要额外的集成包。

2.2 获取 Key 与确认接入信息

先到控制台创建 API Key,建议按项目命名,方便后续轮换和排查。

  • 控制台入口:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=console
  • API Key 管理:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=api-keys
  • 接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc

API 的基础地址是https://taotoken.net/api,注意这个地址不带任何查询参数,直接作为base_url使用即可。Key 拿到后不要硬编码进代码,后面我会用环境变量和配置文件两种方式管理。

注意:Key 属于敏感凭证,提交到 Git 仓库前务必确认已在.gitignore中排除.env和本地配置文件。

2.3 安装依赖

LangChain 拆包比较细,核心库和 OpenAI 兼容集成包要分开装。建议在虚拟环境里操作,避免污染全局。

python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -U langchain langchain-openai python-dotenv pydantic

langchain提供 PromptTemplate、OutputParser 和 LCEL 运行时,langchain-openai提供 ChatOpenAI 这个模型封装,python-dotenv用来读环境变量,pydantic用于定义结构化输出的数据模型。装完后可以用pip show langchain确认版本,避免装到过老的包。

3. 可复制配置:config 骨架与 settings.json

3.1 目录结构

我习惯把配置和业务代码分开,这样换环境时只动配置目录。一个够用的骨架如下:

llm-app/ ├── config/ │ ├── settings.json │ └── loader.py ├── chains/ │ └── review_chain.py ├── .env └── main.py

3.2 settings.json 配置片段

把模型参数、通道地址、默认模型名都放进 JSON,代码里只读不写死。下面这份可以直接用:

{ "provider": { "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "timeout": 60, "max_retries": 2 }, "models": { "default": "gpt-4o-mini", "fast": "gpt-4o-mini", "strong": "gpt-4o" }, "generation": { "temperature": 0.3, "max_tokens": 1024 } }

这里api_key_env存的是环境变量名而不是 Key 本身,代码运行时再去读环境变量,这样配置文件可以安全地进版本库。.env里只写一行:

TAOTOKEN_API_KEY=你的Key

3.3 配置加载器

config/loader.py负责把 JSON 读成字典,并做一次基础校验,避免 Key 缺失时跑到一半才报错:

import json import os from pathlib import Path from dotenv import load_dotenv load_dotenv() CONFIG_PATH = Path(__file__).parent / "settings.json" def load_settings() -> dict: with open(CONFIG_PATH, "r", encoding="utf-8") as f: cfg = json.load(f) env_name = cfg["provider"]["api_key_env"] api_key = os.getenv(env_name) if not api_key: raise RuntimeError(f"环境变量 {env_name} 未设置,请检查 .env 文件") cfg["provider"]["api_key"] = api_key return cfg if __name__ == "__main__": s = load_settings() print("base_url:", s["provider"]["base_url"]) print("default model:", s["models"]["default"])

运行python config/loader.py,如果打印出 base_url 和模型名,说明配置链路是通的。这一步看着简单,但能帮你把「Key 没读到」这类低级问题提前挡掉。

4. Model I/O 三件套:PromptTemplate、ChatModel、OutputParser

4.1 用 ChatOpenAI 接入统一通道

有了配置,模型初始化就三行。注意base_url指向 TaoToken 的 API 地址,model从配置里取:

from langchain_openai import ChatOpenAI from config.loader import load_settings settings = load_settings() provider = settings["provider"] gen = settings["generation"] llm = ChatOpenAI( model=settings["models"]["default"], api_key=provider["api_key"], base_url=provider["base_url"], temperature=gen["temperature"], max_tokens=gen["max_tokens"], timeout=provider["timeout"], max_retries=provider["max_retries"], ) resp = llm.invoke("用一句话说明什么是大模型应用开发框架") print(resp.content)

invoke是最基础的同步调用,返回的是一个 AIMessage 对象,正文在.content里。LangChain 还提供stream做流式、batch做批量、ainvoke做异步,接口签名一致,切换成本很低。

4.2 PromptTemplate:把提示词变成可维护的模板

字符串拼接在提示词简单时没问题,一旦包含角色、背景、格式要求、示例,就会变成一坨。PromptTemplate 用{变量名}占位,把提示词和运行时数据解耦:

from langchain_core.prompts import PromptTemplate review_prompt = PromptTemplate( input_variables=["product", "review"], template=( "你是一位电商运营分析师。\n" "请分析以下关于「{product}」的用户评论," "判断情感倾向并给出改进建议。\n" "评论内容:{review}" ), ) text = review_prompt.format(product="无线耳机", review="音质不错,但续航太短了") print(text)

对话场景更适合用ChatPromptTemplate,它按角色组织消息,和 ChatModel 的输入格式天然对齐:

from langchain_core.prompts import ChatPromptTemplate chat_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一位{role},回答要{style}。"), ("human", "{question}"), ]) messages = chat_prompt.format_messages( role="资深后端工程师", style="简洁并给出代码示例", question="Python 里怎么优雅地合并两个字典?", ) print(messages)

4.3 OutputParser:让模型返回结构化数据

模型默认返回自然语言,但业务往往要 JSON。JsonOutputParser 会自动往提示词里注入格式说明,并把模型输出解析成 Python 字典:

from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field class ReviewResult(BaseModel): sentiment: str = Field(description="情感倾向:正面/负面/中立") keywords: list[str] = Field(description="三个关键词") suggestion: str = Field(description="一句改进建议") parser = JsonOutputParser(pydantic_object=ReviewResult) print(parser.get_format_instructions())

get_format_instructions()返回的是一段格式说明文本,把它塞进提示词,模型就知道该按什么结构输出。如果只需要纯文本,用StrOutputParser即可,它基本是原样透传。

5. LCEL 串联与端到端验证

5.1 用管道符把组件串起来

LCEL 的核心就是|。左边组件的输出作为右边组件的输入,整条链可以像函数一样invoke:

from langchain_core.output_parsers import StrOutputParser chain = review_prompt | llm | StrOutputParser() result = chain.invoke({"product": "无线耳机", "review": "音质不错,但续航太短了"}) print(result)

换成结构化输出,把最后的 StrOutputParser 换成 JsonOutputParser 就行:

structured_chain = review_prompt | llm | parser data = structured_chain.invoke({ "product": "无线耳机", "review": "音质不错,但续航太短了,客服态度也一般" }) print(type(data), data)

5.2 一次完整的端到端验证

把上面的片段整合成main.py,跑一次完整链路。这段代码同时验证了配置加载、模型接入、模板渲染、LCEL 串联和输出解析五个环节:

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from config.loader import load_settings class Analysis(BaseModel): sentiment: str = Field(description="情感倾向") keywords: list[str] = Field(description="关键词列表") summary: str = Field(description="一句话总结") def build_chain(): settings = load_settings() provider = settings["provider"] llm = ChatOpenAI( model=settings["models"]["default"], api_key=provider["api_key"], base_url=provider["base_url"], temperature=0.2, ) parser = JsonOutputParser(pydantic_object=Analysis) prompt = ChatPromptTemplate.from_messages([ ("system", "你是文本分析助手,只输出 JSON。"), ("human", "分析以下评论:{review}\n\n{format}"), ]).partial(format=parser.get_format_instructions()) return prompt | llm | parser if __name__ == "__main__": chain = build_chain() out = chain.invoke({"review": "包装很用心,物流快,就是价格偏高。"}) print(out) assert "sentiment" in out and "keywords" in out print("端到端验证通过")

预期输出是一个字典,包含 sentiment、keywords、summary 三个字段,最后打印「端到端验证通过」。如果这一步成功,说明你的 Model I/O 链路已经完全打通,后面接 RAG、Agent 都只是在这条链上继续加组件。

5.3 流式输出

对话类应用通常要边生成边显示,LCEL 链同样支持stream:

stream_chain = chat_prompt | llm | StrOutputParser() for chunk in stream_chain.stream({ "role": "科普作者", "style": "通俗易懂", "question": "解释一下什么是向量数据库" }): print(chunk, end="", flush=True)

6. 本篇常见报错排查

6.1 401 或鉴权失败

最常见的原因是环境变量没读到。先确认.env和loader.py在同一工作目录下被执行,再确认api_key_env里的变量名和.env中的完全一致(大小写敏感)。如果 Key 是从控制台复制的,注意别把首尾空格带进去。

6.2 404 或模型不存在

多半是base_url写错了。TaoToken 的 API 地址是https://taotoken.net/api,不要在后面手动拼/v1或/chat/completions,LangChain 的 OpenAI 集成会自动补路径。模型名也要和通道支持的名称一致,写错会直接返回模型不存在。

6.3 JSON 解析失败

JsonOutputParser 报解析错误,通常是模型输出里混了 Markdown 代码块标记,比如 ```json。解决办法有两个:一是在 system 提示里明确「只输出 JSON,不要加代码块标记」;二是换用更强的模型,弱模型在长提示下更容易跑偏。另外temperature调低到 0.2 以下也能明显降低格式漂移。

6.4 导入路径报错

LangChain v1.x 把很多类挪到了langchain_core。如果你写from langchain.prompts import PromptTemplate报错,改成from langchain_core.prompts import PromptTemplate。模型类统一从langchain_openai导入,解析器从langchain_core.output_parsers导入,记住这个规律能省不少查文档的时间。

6.5 超时或连接中断

长文本生成容易触发超时。在 ChatOpenAI 里显式设置timeout和max_retries,配置里已经预留了这两个字段。如果批量调用频繁失败,把batch的并发降下来,或者改用异步abatch配合信号量控制并发数。

7. 下一步:把链路接到真实业务

链路跑通之后,你可以按需扩展。想验证不同模型在同一提示下的表现差异,可以直接在模型对话页里对比输出,省去改代码的来回:https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=chat

如果你打算把这条链用到长期编码或 Agent 场景,需要更稳定的额度和更完整的调用能力,可以了解 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding-plan

接入过程中遇到报错,优先翻接入文档里的错误码说明,大部分问题都能对上号:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc

我自己的习惯是,每加一个新组件就先写一个最小invoke验证,确认输入输出格式对得上,再往链里塞。Model I/O 这条链是所有上层能力的地基,地基稳了,后面接检索、接工具、接记忆都不会太痛苦。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询