AI Agent开发学习系列 - langchain之LCEL(3)：Prompt+LLM

基本构成

在 LangChain 的 LCEL（LangChain Expression Language）中，最基本的构成是：
PromptTemplate / ChatPromptTemplate → LLM / ChatModel → OutputParser

• PromptTemplate / ChatPromptTemplate：用于构建和管理提示词（Prompt），可以灵活地插入变量，适配不同的场景。
• LLM / ChatModel：大语言模型（如混元、OpenAI GPT等），负责根据提示词生成内容。
• OutputParser：输出解析器，用于将模型的原始输出转换为你需要的格式（如字符串、JSON、结构化数据等）。

这三者串联起来，就形成了最基础的链式调用结构。例如：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import SecretStr
import os
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model="hunyuan-lite",  # 模型名称
    temperature=0
)

prompt = ChatPromptTemplate.from_template("请讲一个关于{topic}的故事。")

chain = prompt | llm

chain.invoke({"topic": "月亮"})

AIMessage(content='从前，有一个古老的村庄，村子里的居民们相信月亮是他们的守护神。每天夜晚，他们都会仰望天空，看着月亮在天空中越升越高，感到无比的宁静和安详。\n\n有一天晚上，村里来了一位神秘的老人。他告诉村民们，月亮其实是一个被诅咒的公主，她的灵魂被困在了地球上。只有找到一位勇敢的年轻人，才能解除这个诅咒，让月亮恢复自由。\n\n村子里有一个名叫小明的年轻人，他勇敢、善良，决定去寻找解除诅咒的方法。他告别了家人和朋友，踏上了漫长的旅程。\n\n在旅途中，小明遇到了各种困难和挑战，但他从未放弃过。他攀爬陡峭的山峰，穿越茂密的森林，横渡汹涌的河流。最终，他来到了一个神秘的岛屿。\n\n在岛屿的中心，小明发现了一个古老的祭坛，祭坛上刻着解除诅咒的仪式。他按照仪式的要求，献上了自己带来的鲜花、果实和宝石，向月亮祈祷。\n\n突然间，天空中传来了一阵美妙的音乐，月亮的光芒变得更加明亮。一位美丽的仙女出现在小明面前，她告诉小明，他已经成功地解除了诅咒，现在月亮已经恢复了自由。\n\n小明带着仙女回到村子，村民们欢欣鼓舞，感激不已。从那以后，村子里的居民们更加尊敬和爱护月亮，每晚都会仰望天空，感谢这位勇敢的年轻人带来的和平与宁静。\n\n而小明也成了村子的英雄，他的勇敢和善良被传颂了一代又一代。从此，村子与月亮和谐共处，共同守护着这片美丽的土地。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 336, 'prompt_tokens': 9, 'total_tokens': 345, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'hunyuan-lite', 'system_fingerprint': '', 'id': '1f26acbbaf75a893ba12b8d3c602c95f', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--bcff3a64-22e2-41c8-933b-b10a01b912bb-0', usage_metadata={'input_tokens': 9, 'output_tokens': 336, 'total_tokens': 345, 'input_token_details': {}, 'output_token_details': {}})

这样就实现了“输入主题→生成故事→输出解析”的完整流程。

自定义停止输出符

在 LangChain 的 LCEL 中，bind 方法用于为模型绑定额外的参数，这些参数会在模型调用时自动应用。

chain = prompt | llm.bind(stop=["。"])

chain.invoke({"topic": "月亮"})

输出

AIMessage(content='从前，有一个古老的村庄，村子里的居民们相信月亮是他们的守护神。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 9, 'total_tokens': 26, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'hunyuan-lite', 'system_fingerprint': '', 'id': 'b5f1b82e961e3470e675c99530552034', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--532ba994-3a70-488e-8a2e-791dffbbcea6-0', usage_metadata={'input_tokens': 9, 'output_tokens': 17, 'total_tokens': 26, 'input_token_details': {}, 'output_token_details': {}})

兼容openai函数调用的方式

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import SecretStr
import os
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model="hunyuan-lite",  # 模型名称
    temperature=0
)

prompt = ChatPromptTemplate.from_template(
    "请用如下JSON格式输出故事：{{\"setup\": \"...\", \"punchline\": \"...\"}}，主题是：{topic}"
)

functions = [
    {
        "name": "story",
        "description": "讲故事",
        "parameters": {
            "type": "object",
            "properties": {
                "setup": {"type": "string", "description": "故事主题"},
                "punchline": {"type": "string", "description": "故事结尾"}
            },
            "required": ["setup", "punchline"],
        },
    }
]

chain = prompt | llm.bind(function_call={"name": "story"}, functions=functions)

chain.invoke({"topic": "月亮"}, config={})

输出

AIMessage(content='{\n  "setup": "在一个宁静的夜晚，小明坐在窗前，凝视着天空中闪烁的星星。突然，他注意到了一颗异常明亮的月亮。",\n  "punchline": "小明惊讶地发现，这颗明亮的月亮竟然会随着他的心跳而移动！"\n}', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 66, 'prompt_tokens': 27, 'total_tokens': 93, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'hunyuan-lite', 'system_fingerprint': '', 'id': 'f16351bbd58ce16fe8f2400906e00be3', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--51b9f5f2-5b56-4eba-9fa5-b3ff69f914e0-0', usage_metadata={'input_tokens': 27, 'output_tokens': 66, 'total_tokens': 93, 'input_token_details': {}, 'output_token_details': {}})

输出解析器

from langchain_core.output_parsers import StrOutputParser

chain = prompt | llm | StrOutputParser()

chain.invoke({"topic": "月亮"})

输出

'{\n  "setup": "在一个宁静的夜晚，小明坐在窗前，凝视着天空中闪烁的星星。突然，他注意到了一颗异常明亮的月亮。",\n  "punchline": "小明惊讶地发现，这颗明亮的月亮竟然会随着他的心跳而移动！"\n}'

使用Runnables来连接多链结构

from operator import itemgetter

from langchain.schema import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import SecretStr
import os
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model="hunyuan-lite",  # 模型名称
    temperature=0
)

prompt1 = ChatPromptTemplate.from_template("{who}是哪国人？")
prompt2 = ChatPromptTemplate.from_template(
    "{country}在哪个洲？用{language}回答。"
)

chain1 = prompt1 | llm | StrOutputParser()
chain2 = (
    {"country": chain1, "language": itemgetter("language")} 
    | prompt2 | llm | StrOutputParser()
)

chain2.invoke({"who": "姚明", "language": "英文"})

输出

'姚明是中国人，中国位于亚洲。所以答案是：Asia.'

from langchain_core.runnables import RunnablePassthrough

prompt1 = ChatPromptTemplate.from_template(
    "生成一个{attribute}属性的颜色。除了返回这个颜色的名字不要做其他事："
)

prompt2 = ChatPromptTemplate.from_template(
    "什么水果是这个颜色:{color}，只返回这个水果的名字不要做其他事情："
)

prompt3 = ChatPromptTemplate.from_template(
    "哪个国家的国旗有这个颜色:{color}，只返回这个国家的名字不要做其他事情："
)

prompt4 = ChatPromptTemplate.from_template(
    "有这个颜色的水果是{fruit}，有这个颜色的国旗是{country}？"
)

model_parser = llm | StrOutputParser()
# 生成一个颜色
color_generator = (
    {"attribute": RunnablePassthrough()} | prompt1 | {"color": model_parser}
)

color_to_fruit = prompt2 | model_parser
color_to_country = prompt3 | model_parser

question_generator = (
    color_generator | {"fruit": color_to_fruit, "country": color_to_country} | prompt4
)

question_generator.invoke("强烈的")

输出

ChatPromptValue(messages=[HumanMessage(content='有这个颜色的水果是苹果，有这个颜色的国旗是俄罗斯？', additional_kwargs={}, response_metadata={})])

示例：唯物辩证链

from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from operator import itemgetter
from pydantic import SecretStr
import os
from dotenv import load_dotenv

llm = ChatOpenAI(
    api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model="hunyuan-lite",  # 模型名称
    temperature=0
)

planner = (
    ChatPromptTemplate.from_template("生成一个关于{input}的论点")
    | llm
    | StrOutputParser()
    | {"base_response": RunnablePassthrough()}
)

arguments_for = (
    ChatPromptTemplate.from_template(
        "列出以下内容的优点或积极方面:{base_response}"
    )
    | llm
    | StrOutputParser()
)

arguments_against = (
    ChatPromptTemplate.from_template(
        "列出以下内容的缺点或消极方面:{base_response}"
    )
    | llm
    | StrOutputParser()
)

final_responder = (
    ChatPromptTemplate.from_messages(
        [
            ("system", "根据评论生成最终的回复"),
            ("ai", "{original_response}"),
            ("human", "积极:\n{results_1}\n\n消极:\n{results_2}"),
        ]
    )
    | llm
    | StrOutputParser()
)

chain = (
    planner
    | {
        "results_1": arguments_for,
        "results_2": arguments_against,
        "original_response": itemgetter("base_response"),
    }
    | final_responder
)

chain.invoke({"input": "AI取代人类"})

输出

'AI取代人类的论点的优缺点分析如下：\n\n优点：\n1. **提高效率和生产力**：AI能够自动化重复性、繁琐或危险的任务，从而显著提升工作效率和生产力。\n2. **减少人为错误**：AI系统不受人类情绪、疲劳或认知偏差的影响，执行任务时更加准确可靠。\n3. **解决劳动力短缺问题**：AI的引入有助于填补行业空缺，特别是在危险或不适合人类工作的环境中。\n4. **成本效益**：从长远来看，AI取代人类可以降低企业的运营成本。\n\n缺点：\n1. **就业市场变化**：AI技术的发展可能导致一些传统职业被机器取代，从而使部分劳动力失业。\n2. **伦理和社会问题**：AI的决策过程可能受到算法和数据的影响，引发伦理和社会问题。\n3. **技术局限性**：AI在处理复杂问题、理解人类情感和创造力方面的能力有限。\n4. **人机交互问题**：AI技术在自然语言理解、情感识别等方面仍存在不足。\n5. **经济不平等**：AI技术的广泛应用可能导致经济不平等问题的加剧。\n6. **道德责任归属**：当AI系统出现错误或造成损害时，确定道德责任归属可能变得复杂。\n7. **技术失控风险**：存在一种担忧，即AI系统可能最终失控。\n\n综合以上分析，AI取代人类的论点既有其积极的一面，也伴随着一系列挑战和问题。在推进AI技术发展的同时，需要全面考虑这些因素，制定合理的政策和措施，以确保技术的可持续发展和社会的和谐进步。'

自定义输出解析器

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import (
    ChatPromptTemplate,
)
from langchain_experimental.utilities import PythonREPL
from langchain_openai import ChatOpenAI
from pydantic import SecretStr
import os
from dotenv import load_dotenv

load_dotenv()


template = """根据用户需求帮助用户编写python代码。

只需要返回makedown格式的python代码，比如：

```python
...
```"""

prompt = ChatPromptTemplate.from_messages([("system", template), ("human", "{input}")])

model = ChatOpenAI(
    api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model="hunyuan-lite",  # 模型名称
    temperature=0
)

# 自定义输出解析，只返回python代码
def _sanitize_output(text: str):
    _, after = text.split("```python")
    return after.split("```")[0]

# 定义链
chain = prompt | model | StrOutputParser() | _sanitize_output | PythonREPL().run

chain.invoke({"input": "计算1到100的和"})

Python REPL can execute arbitrary code. Use with caution.
'5050\n'