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

AI大模型（六）Langchain核心模块与实战（一）

news 2025/7/6 16:48:21

LangChain调用

在这里插入图片描述

采用LangChain调用LLM

import osfrom langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAIos.environ['LANGCHAIN_TRACING_V2'] = "true"
os.environ['LANGCHAIN_API_KEY'] = "xxx"# 调用大预言模型的三件事情
# 1.创建模型
model = ChatOpenAI(model='gpt-4-turbo',api_key='sk-xx',base_url='xxx')# 2. 给出提示prompt
msg = [SystemMessage(content='请将以下的内容翻译成意大利语'),HumanMessage(content='你好，请问你要去哪里？')
]# result = model.invoke(msg)
# print(result)# 简单的解析响应数据
#3. 创建返回数据的解析器
parser = StrOutputParser()
# print(parser.invoke(result))#4. 得到链
chain = model | parser#5.直接使用chain来调用
print(chain.invoke(msg))

在这里插入图片描述

Langchain的提示模板

import osfrom langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAIos.environ['LANGCHAIN_TRACING_V2'] = "true"
os.environ['LANGCHAIN_API_KEY'] = "xxx"# 调用大预言模型的三件事情
# 1.创建模型
model = ChatOpenAI(model='gpt-4-turbo',api_key='sk-xx',base_url='xxx')# 2. 给出提示prompt
msg = [SystemMessage(content='请将以下的内容翻译成意大利语'),HumanMessage(content='你好，请问你要去哪里？')
]# result = model.invoke(msg)
# print(result)# 简单的解析响应数据
#3. 创建返回数据的解析器
parser = StrOutputParser()
# print(parser.invoke(result))# 定义提示模板
prompt_template = ChatPromptTemplate.from_messages([('system','请将下面的内容翻译成{language}'),('user',"{text}")])#4. 得到链
chain = prompt_template | model | parser#5.直接使用chain来调用
# print(chain.invoke(msg))
print(chain.invoke({'language':'English','text':'我下午还有一节课，不能去打球了'}))

得到如下输出结果
在这里插入图片描述

部署你的LangChian程序

执行安装命令：
pip install “langserve[all]”

import osfrom fastapi import FastAPI
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langserve import add_routesos.environ['LANGCHAIN_TRACING_V2'] = "true"
os.environ['LANGCHAIN_API_KEY'] = "xxx"# 调用大预言模型的三件事情
# 1.创建模型
model = ChatOpenAI(model='gpt-4-turbo',api_key='xxx',base_url='https://xiaoai.plus/v1')# 简单的解析响应数据
#3. 创建返回数据的解析器
parser = StrOutputParser()
# print(parser.invoke(result))# 定义提示模板
prompt_template = ChatPromptTemplate.from_messages([('system','请将下面的内容翻译成{language}'),('user',"{text}")])#4. 得到链
chain = prompt_template | model | parser#5.直接使用chain来调用
# print(chain.invoke(msg))
print(chain.invoke({'language':'English','text':'我下午还有一节课，不能去打球了'}))# 把我们的程序部署成服务
# 创建fastAPI应用
app = FastAPI(title='我的Langchain服务',version='V1.0',description='使用Langchain翻译任何语句的服务器')
add_routes(app,chain,path='/chainDemo'
)
if __name__ == "__main__":import uvicornuvicorn.run(app, host="localhost", port=8000)

在这里插入图片描述
在 FastAPI 中使用 LangServe 的 add_routes 函数注册 LangChain 链时，端点路径的设计遵循了 LangServe 的特定约定。以下是详细解释：

原因分析：
RESTful 端点设计：
LangServe 为每个链注册一组标准化的 REST 端点，而非单一端点。这些端点包括：

POST /your_path/invoke：同步调用链

POST /your_path/batch：批量调用

POST /your_path/stream：流式输出

GET /your_path/input_schema：获取输入 JSON Schema

GET /your_path/output_schema：获取输出 JSON Schema

通过以上学习，未来如果我们写好了一个大预言模型的真正的消费者级别的产品，我们可以把这个产品部署成服务的方式，然后提供给其他的应用程序或者人去调用

Langchain构建聊天机器人

这个聊天机器人能够进行对话并记住之前的互动。
需要安装：pip install langchain_community

Chat History：它允许聊天机器人“记住”过去的互动，并在回应后续问题时考虑它们。
流试输出

import osfrom langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_community.chat_message_histories import  ChatMessageHistoryos.environ['LANGCHAIN_TRACING_V2'] = "true"
os.environ['LANGCHAIN_PROJECT'] ="LangchainDemo"
os.environ['LANGCHAIN_API_KEY'] = "xxx"# 聊天机器人案例
# 1.创建模型
model = ChatOpenAI(model='gpt-4-turbo',api_key='sk-xxx',base_url='https://xiaoai.plus/v1')# 定义提示模板prompt_template = ChatPromptTemplate.from_messages([('system','你是一个乐于助人的助手。用{language}尽你所能回答所有问题。'),MessagesPlaceholder(variable_name='my_msg')
])#4. 得到链
chain = prompt_template | model# 开始聊天对话，保存聊天记录
store = {} #所有用户的聊天记录都保存到store。 key：sessionId, value:历史聊推荐记录对象# 此函数预期将接收一个session_id并返回一个消息历史记录对象。
def get_session_history(session_id: str):if session_id not in store:store[session_id] = ChatMessageHistory()return store[session_id]do_message = RunnableWithMessageHistory(chain,get_session_history,input_messages_key='my_msg'  # 每次聊天时候发送msg的key
)config = {'configurable': {'session_id': 'zs1234'}}  # 给当前会话定义一个sessionId
# 第一轮聊天
resp = do_message.invoke({'my_msg':[HumanMessage(content='你好啊！我是LaoLiu')],'language':'中文',
},config = config
)
print(resp.content)# 第二轮聊天（历史聊天的体现）
resp2 = do_message.invoke({'my_msg':[HumanMessage(content='请问：我的名字是什么')],'language':'中文',
},config = config
)
print(resp2.content)# 第三轮：返回的数据是流式的
for resp in do_message.stream({'my_msg':[HumanMessage(content='请问：请给我讲一个笑话')],'language':'English',
},config = config
):# 每次resp都是一个tokenprint(resp.content,end='-')

在这里插入图片描述
store里面存储的内容：

{'zs1234': InMemoryChatMessageHistory(messages=[HumanMessage(content='你好啊！我是LaoLiu', additional_kwargs={}, response_metadata={}), AIMessage(content='你好，LaoLiu！很高兴见到你。有什么我可以帮助你的吗？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 32, 'prompt_tokens': 50, 'total_tokens': 82, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-4-turbo', 'system_fingerprint': 'fp_5603ee5e2e', 'id': 'chatcmpl-GnfWBMpGjL0ysDrC8RooLYwQ8rcCQ', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--85c7f74a-39fd-41f0-bc7e-c07fa3a2f6d7-0', usage_metadata={'input_tokens': 50, 'output_tokens': 32, 'total_tokens': 82, 'input_token_details': {}, 'output_token_details': {}}), HumanMessage(content='请问：我的名字是什么', additional_kwargs={}, response_metadata={}), AIMessage(content='你刚刚告诉我，你的名字是LaoLiu。需要帮助解决其他问题吗？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 33, 'prompt_tokens': 100, 'total_tokens': 133, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-4-turbo', 'system_fingerprint': 'fp_5603ee5e2e', 'id': 'chatcmpl-j0W7QTq3XlqPD6QHYz20RzpaFCZH2', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--9f7da6e0-af80-414a-bdfb-082ef055d931-0', usage_metadata={'input_tokens': 100, 'output_tokens': 33, 'total_tokens': 133, 'input_token_details': {}, 'output_token_details': {}}), HumanMessage(content='请问：请给我讲一个笑话', additional_kwargs={}, response_metadata={}), AIMessageChunk(content="Sure! Here's a light joke for you:\n\nWhy don't skeletons fight each other?\n\nThey don't have the guts. \n\nI hope that brings a smile to your face!", additional_kwargs={}, response_metadata={'finish_reason': 'stop', 'model_name': 'gpt-4-turbo', 'system_fingerprint': 'fp_5603ee5e2e'}, id='run--f50c092c-8509-452d-97ed-9af20b7fbb98')])}