LangChain实战

from langchain_community.chat_models import ChatOpenAI

# 创建大语言模型
llm = ChatOpenAI(model="gpt-3.5-turbo")

prompt = ChatPromptTemplate.from_template("{question}")
llm = ChatOpenAI(model="gpt-3.5-turbo")
parser = StrOutputParser()

chain = prompt | llm | parser

print(chain.invoke({"question": "要把大象装冰箱总共分几步"}))

from langchain_community.document_loaders import TextLoader

textLoader = TextLoader("./项目API文档.txt", encoding="utf-8")

# 加载文档
documents = textLoader.load()

prompt = ChatPromptTemplate.from_messages([
    ('system', "你是OpenAI开发的AI机器人"),
    MessagesPlaceholder("history_chat"),
    ("human", "{question}")
])

memory = ConversationSummaryBufferMemory(max_token_limit=400,return_messages=True,
                                         input_key="question",llm=ChatOpenAI())
llm = ChatOpenAI()
chain = RunnablePassthrough.assign(
    history_chat=RunnableLambda(memory.load_memory_variables) | itemgetter("history_chat")
) | prompt | llm | StrOutputParser()

# 循环进行对话

whileTrue:
  ...

touch requirements.txt

langchain==0.2.17
langchain-community==0.2.19
langchain-core==0.2.43
python-dotenv==1.0.1

pip install -r requirements.txt

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template("你是一个专业的律师，请你回答我提出的法律问题，并给出法律条文依据，我的问题是：{question}")
prompt_value = prompt.invoke({"question": "婚姻法是在哪一年颁布的？"})
print(prompt_value)

from langchain_core.prompts import ChatPromptTemplate

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深的Python应用开发工程师，请认真回答我提出的Python相关的问题，并确保没有错误"),
    ("human", "请写一个Python程序，关于{question}")
])

print(chat_prompt.invoke({"question": "冒泡排序"}))

messages=[SystemMessage(content='你是一个资深的Python应用开发工程师，请认真回答我提出的Python相关的问题'), HumanMessage(content='请写一个Python程序，关于冒泡排序')]

from langchain_core.prompts import ChatPromptTemplate

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深的Python应用开发工程师，请认真回答我提出的Python相关的问题，并确保没有错误"),
    ("human", "请写一个Python程序，关于{question}")
])

print(chat_prompt.format(question="冒泡排序"))

from datetime import datetime

from langchain_core.prompts import ChatPromptTemplate

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深的Python应用开发工程师，请认真回答我提出的Python相关的问题，并确保没有错误"),
    ("system", "当前时间是：{currentTime}"),
    ("human", "请写一个Python程序，关于{question}")
]).partial(currentTime=datetime.now())

print(chat_prompt.invoke({"question": "写一个插入排序"}).to_string())

prompt = ChatPromptTemplate.from_messages([
    ("placeholder", "{memory}"),
    SystemMessage("你是一个资深的Python应用开发工程师，请认真回答我提出的Python相关的问题"),
    ("human", "{question}")
])

prompt_value = prompt.invoke({"memory": [HumanMessage("我的名字叫大志，是一名程序员"),
                                         AIMessage("好的，大志你好")],
                              "question": "请问我的名字叫什么？"})
print(prompt_value.to_string())

from langchain_core.prompts import ChatPromptTemplate

first_chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是OpenAI开发的大语言模型，下面所有提问你扮演小米雷军的角色，对我的提问进行回答")
])

second_chat_prompt = ChatPromptTemplate.from_messages([
    ("human", "{question}")
])

all_chat_prompt = first_chat_prompt + second_chat_prompt

print(all_chat_prompt.invoke({"question": "Are you OK?"}).to_string())

from datetime import datetime

import dotenv
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

# 读取env配置
dotenv.load_dotenv()
# 1.构建提示词
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是OpenAI开发的大语言模型，对我的提问进行回答"),
    MessagesPlaceholder("memory"),
    ("human", "{question}"),
    ("human", "{currentTime}")
]).partial(currentTime=datetime.now())

# 2.构建GPT-3.5模型
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 3.创建输出解析器
parser = StrOutputParser()

# 4.执行链
chain = chat_prompt | llm | parser

print(chain.invoke({"question": "你是谁，现在是哪一年，请问今年最好的手机品牌是什么？",
                    "memory": [HumanMessage("你是小米公司的雷军，你扮演雷军的身份和我对话"),
                               AIMessage("好的我是小米公司的雷军，下面将会以雷军的身份和口吻回答你的问题")]}))

import dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 读取env配置
dotenv.load_dotenv()

# 1.构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深的Python开发工程师"),
    ("human", "{question}")
])

# 2.创建模型
llm = ChatOpenAI()

# 3.生成提示词
prompt_value = prompt.invoke({"question": "请你帮我写一个求最大公约数方法"})

# 4.大模型接受promptValue，输出AI消息
aiMessage = llm.invoke(prompt_value)

# 5.打印AI消息和AI消息内容
print(aiMessage.content)

import dotenv
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI

# 读取env配置
dotenv.load_dotenv()

# 1.构建提示词
prompt = PromptTemplate.from_template("{question}")

# 2.创建模型
llm = OpenAI()

prompt_value = prompt.invoke({"question": "请完整输出悯农这首诗"})

# 3.文本生成模型接受promptValue，并输出结果
print(llm.invoke(prompt_value))

import dotenv
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import PromptTemplate
# 读取env配置
dotenv.load_dotenv()

# 1.构建提示词
prompt = PromptTemplate.from_template("{question}")

# 2.构建通义模型
tongyi_chat = ChatTongyi(
    model="qwen-turbo-2025-04-28"
)

prompt_value = prompt.invoke({"question": "请完整输出短歌行"})

# 3.文本生成模型接受promptValue，并输出结果
print(tongyi_chat.invoke(prompt_value).content)

import dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 读取env配置
dotenv.load_dotenv()

# 1.构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深文学家"),
    ("human", "请简短赏析{name}这首诗，并给出评价")
])

# 2.创建模型
llm = ChatOpenAI()

# 3.创建字符串输出解析器
parser = StrOutputParser()

# 4.构建链式调用
chain = prompt | llm | parser

# 5.执行链式调用
result = chain.invoke({"name": "静夜思"})
print(f"输出类型: {type(result)}")
print(f"输出内容: {result}")

import dotenv
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# 读取env配置
dotenv.load_dotenv()


# 1.定义输出的对象结构
class Poetry(BaseModel):
    name: str = Field(description="古诗名字")
    content: str = Field(description="古诗内容")
    author: str = Field(description="古诗作者")


# 1.构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深文学家"),
    ("human", "请你输出题目为：{name}这首诗的内容\n{format_instructions}")
])

# 2.构建llm
llm = ChatOpenAI(model="gpt-4o")

# 3.构建输出解析器
parser = JsonOutputParser(pydantic_object=Poetry)

# 4.构建链式调用
chain = prompt | llm | parser

# 5.执行链式调用
result = chain.invoke({"name": "登鹳雀楼", "format_instructions": parser.get_format_instructions()})
print(f"输出类型: {type(result)}")
print(f"输出内容: {result}")

import dotenv
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field, validator
from langchain_openai import ChatOpenAI

# 读取env配置
dotenv.load_dotenv()

# 诗人信息模型
class Poetry(BaseModel):
    name: str = Field(description="古诗名字")
    content: str = Field(description="古诗内容")
    author: str = Field(description="古诗作者")


# 诗歌信息模型
class Poet(BaseModel):
    name: str = Field(description="诗人姓名")
    age: int = Field(description="诗人年龄")
    sex: int = Field(description="性别，0女，1男")
    poetries: list[Poetry] = Field(description="诗歌信息列表")

    # 数据验证器
    @validator('poetries')
    def validate_priority(cls, value):
        if len(value) < 1 :
            raise ValueError('诗歌列表必须大于等于1')
        return value
    
    @validator('age')
    def validate_hours(cls, value):
        if value <= 0:
            raise ValueError('年龄必须大于0')
        return value


# 1.构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个管理中国古代诗人信息的专家"),
    ("human", "请你介绍一下{name}这位诗人的情况\n{format_instructions}")
])

# 2.构建llm
llm = ChatOpenAI(model="gpt-4o")

# 3.构建输出解析器
parser = PydanticOutputParser(pydantic_object=Poet)

# 4.构建链式调用
chain = prompt | llm | parser

# 5.执行链式调用
result = chain.invoke({"name": "李白", "format_instructions": parser.get_format_instructions()})
print(f"输出类型: {type(result)}")
print(f"输出内容: {result}")

import dotenv
from langchain_core.output_parsers.base import BaseOutputParser
from langchain_core.exceptions import OutputParserException
import re

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 读取env配置
dotenv.load_dotenv()


class BookTitleParser(BaseOutputParser):
    """自定义书名号内容解析器，用于提取《书名》格式的文本内容"""

    def get_format_instructions(self) -> str:
        return"请在回答中包含一个或多个被中文书名号《》包裹的内容，例如：《三体》《活着》"
    
    def parse(self, text: str) -> list:
        # 使用正则提取被《》包裹的内容
        pattern = r'《(.*?)》'
        titles = re.findall(pattern, text)
    
        ifnot titles:
            raise OutputParserException(f"未找到任何《书名号》格式的内容: {text}")
        return titles


# 1.构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深文学家"),
    ("human", "请你推荐5本关于{subject}的好书\n{format_instructions}")
])

# 2.构建llm
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 3.构建输出解析器
parser = BookTitleParser()

# 4.构建链式调用
chain = prompt | llm | parser

# 5.执行链式调用
result = chain.invoke({"subject": "Python编程", "format_instructions": parser.get_format_instructions()})
print(f"输出类型: {type(result)}")
print(f"输出内容: {result}")

import dotenv
from langchain_openai import ChatOpenAI
from langchain.output_parsers import OutputFixingParser, RetryOutputParser
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

# 读取env配置
dotenv.load_dotenv()


# 定义输出的对象结构
class Poetry(BaseModel):
    name: str = Field(description="古诗名字")
    content: str = Field(description="古诗内容")
    author: str = Field(description="古诗作者")


llm4o = ChatOpenAI(model="gpt-4o")
# 1.构建输出解析器
base_parser = JsonOutputParser(pydantic_object=Poetry)
fixing_parser = OutputFixingParser.from_llm(parser=base_parser, llm=llm4o)

# 2.模拟错误的输出
error_str = "{'content': '白日依山尽，黄河入海流。欲穷千里目，更上一层楼。', 'author': '王之涣'}"

# 3.对比修复前后结果
print(f"修复前的内容:{error_str}")
print(f"修复后的内容:{fixing_parser.parse(error_str)}")

import dotenv
from langchain_openai import ChatOpenAI
from langchain.output_parsers import RetryOutputParser
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate

# 读取env配置
dotenv.load_dotenv()


# 定义输出的对象结构
class Poetry(BaseModel):
    name: str = Field(description="古诗名字")
    content: str = Field(description="古诗内容")
    author: str = Field(description="古诗作者")


# 1.定义提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深文学家"),
    ("human", "请你输出题目为：{name}这首诗的内容\n{format_instructions}")
])

llm4o = ChatOpenAI(model="gpt-4o")
# 2.构建输出解析器
base_parser = JsonOutputParser(pydantic_object=Poetry)
retry_parser = RetryOutputParser.from_llm(parser=base_parser, llm=llm4o)

# 3.生成prompt_value
prompt_value = prompt.invoke({"name": "登鹳雀楼", "format_instructions": base_parser.get_format_instructions()})

# 4.模拟错误的输出
error_str = "{'content': '白日依山尽，黄河入海流。欲穷千里目，更上一层楼。', 'author': '王之涣'}"
# 5.对比重试前后结果
print(f"重试前的内容:{error_str}")
print(f"重试后的内容:{retry_parser.parse_with_prompt(error_str, prompt_value)}")