当前位置：首页 > news >正文

Langchain的LCEL组件

news 2025/11/13 7:11:17

一.提示词模板和提示词模板部分格式化

1.llm提示词模板

# llm模型提示词模板使用
# 提示词字符串
prompt_str="这是一个提示词，解释{text}的含义"
# 提示词模板
prompt = PromptTemplate.from_template(prompt_str)
vars = "langchain"
prompt.format(text=vars)
result=client.invoke(prompt.format(text="vars"))
print(result)

2.chat聊天提示模板

# chat聊天模版
chat_template = ChatPromptTemplate.from_messages([("system", "解释词语含义."),("human", "{text}"),]
)
# 模板变成消息列表，将占位解析为变量
chat_template=chat_template.format_messages(text="老虎")
# invoke只能接受消息列表
print(client.invoke(chat_template))

3.少样本提示词模板

# 示例
examples = [{"sinput":"3+3","soutput":"6","sdescription":"加法运算"},{"sinput":"3*3","soutput":"9","sdescription":"乘法运算"},
]# 配置一个提示模板
examples_prompt_template = "算式:{sinput} 值:{soutput} 类型:{sdescription}"# 少样本提示词模板
few_prompt = FewShotPromptTemplate(# 样例# 示例examples=examples,# 示例提示词模板example_prompt=PromptTemplate.from_template(examples_prompt_template),# 用户# 类似系统提示词，添加到最前面prefix="请根据结果，给出正确的算式类型。",# 用户需要的操作，添加到最后面suffix="现在给你一个算式：{user_input},值：{user_output}",
)chain = few_prompt | client | StrOutputParser()print(chain.invoke({"user_input":"4+4","user_output":"8"}))

二、输出解析器

1.字符串

2.json

提示使用key，value的json格式输出

3.csv

提示使用，分开

4.xml

# chat聊天模版
chat_template = ChatPromptTemplate.from_messages([("system", "解释词语含义."),("human", "{text}"),]
)
# str = StrOutputParser()
# chain = chat_template | client | str
# print(chain.invoke({"text":"老虎"}))
#
# json=JsonOutputParser()
# chain = chat_template | client | json
# print(chain.invoke({"text":"老虎。问题用Q,回答用A,使用json格式输出"}))csv = CommaSeparatedListOutputParser()
# chain = chat_template | client | csv
chain = RunnableSequence(chat_template, client, csv)
print(chain.invoke({"text":"老虎。回答用,分开"}))

三、LCEL高级组件

1.RunnableSequnece=管道符

管道符号的左右两边必须是Runnable对象

如果 | 一边是Runnable对象，另外一边会被自动包装

2.RunnableLambda函数链

自定义函数加入链，但是该函数只能接收一个参数，这个参数可以是字典

管道符号左右需要时一个Runnable对象

@ chain = RunnableLambda(函数)

chat_template = ChatPromptTemplate.from_template(" {len1} + {len2}等于多少？")@chain
def length_function(text):return len(text)
def _multiple_length_function(text1, text2):return len(text1) * len(text2)
@chain
def multiple_length_function(_dict):return _multiple_length_function(_dict["text1"], _dict["text2"])chain2 = ({# "len1": RunnableLambda(lambda x:x["key1"]) | length_function,"len1": itemgetter("key1") | length_function,"len2":({"text1": itemgetter("key1"), "text2": itemgetter("key2")} | multiple_length_function)}| chat_template| client| StrOutputParser()
)
print(chain2.invoke({"key1":"12345","key2":"1234"}))

3.RunnableParallel并行链

@chain
def add_one(x):return x + 1@chain
def add_two(x):return x + 2@chain
def add_three(x):return x + 3# r1 = RunnableLambda(add_one)
# r2 = RunnableLambda(add_two)
# r3 = RunnableLambda(add_three)chain = add_one | RunnableParallel(two=add_two,three=add_three
)
# 返回字典类型
print(chain.invoke(1))

4.RunnablePassthrough.assign传递数据链

.assign()的输入必须是 dict，

# def add_one(d):
#     return d["aa"] + 1
# chain = RunnableParallel(
#     passed=RunnablePassthrough(),
#     pass_two=RunnablePassthrough().assign(key=add_one)
# )chain = RunnableParallel(passed=RunnablePassthrough(),# assign(接收字典)pass_two=RunnablePassthrough().assign(key=lambda x:x["aa"]+1)
)
# 返回字典类型
print(chain.invoke({"aa":1}))

四、链的dbug调试

from langchain_core.globals import set_debug, set_verbose
# 0. 调试开关（任选其一或全开）
# 0-a 控制台实时日志
set_debug(True)          # 最详细：prompt / completion / token / 耗时
set_verbose(True)        # 链级步骤信息