FastAPI、langchain搭建chatbot，langgraph实现汗青记录

情况：openEuler、python 3.11.6、Azure openAi、langchain  0.3.3、langgraph  0.2.38
背景：基于FastAPI、langchain实现一个QA系统，要求实现汗青记录以及存储特征信息
时间：20241022
阐明：在汗青记录的存储中，增加自定义的存储内容，后期还要以流的形式实现
官方文档地点：Add summary of the conversation history
源码地点：尚无
1、情况搭建（略）

2、功能案例

1、实现记录和总结（langgraph）

概要：使用langgraph实现对话系统，该系统具备谈天汗青记录和总结功能
官方文档地点：Add summary of the conversation history
 实现具有总结和汗青记录功能的chatbot

1. from typing import Literal
3. from langchain_core.messages import SystemMessage, RemoveMessage
4. from langgraph.checkpoint.memory import MemorySaver
5. from langgraph.graph import MessagesState, StateGraph, START, END
6. from langchain_core.messages import HumanMessage
7. from langchain_openai import AzureChatOpenAI
9. # 加载环境中的api_key等信息
10. from dotenv import load_dotenv
12. load_dotenv(".env")
14. # 创建memory，用于记录state信息
15. memory = MemorySaver()
17. # 在state中新增summary参数
18. class State(MessagesState):
19.     summary: str
21. # 使用Azure作为chain
22. model = AzureChatOpenAI(deployment_name="aicontent-validation")
24. # 定义访问模型的逻辑，这是node
25. def call_model(state: State):
26.     # summary是否存在，影响下一次的回答内容
27.     summary = state.get("summary", "")
28.     if summary:
29.         system_message = f"Summary of conversation earlier: {summary}"
30.         messages = [SystemMessage(content=system_message)] + state["messages"]
31.     else:
32.         messages = state["messages"]
33.     response = model.invoke(messages)
34.     # 将内容以列表的形式返回，会添加到state的列表中
35.     return {"messages": [response]}
37. # 确认结束还是执行summarize_conversation
38. def should_continue(state: State) -> Literal["summarize_conversation", END]:
39.     """Return the next node to execute."""
40.     messages = state["messages"]
41.     # 如果对话内容超过5条，则进行总结（转向执行summarize_conversation）
42.     if len(messages) > 5:
43.         return "summarize_conversation"
44.     # 这个graph结束了
45.     return END
47. def summarize_conversation(state: State):
48.     # 在state中获取summary的内容，没有则定义为空字符串
49.     summary = state.get("summary", "")
50.     if summary:
51.         # summary_message可以认为是一个prompt，是否存在summary_message，直接影响总结
52.         summary_message = (
53.             f"This is summary of the conversation to date: {summary}\n\n"
54.             "Extend the summary by taking into account the new messages above:"
55.         )
56.     else:
57.         summary_message = "Create a summary of the conversation above:"
59.     messages = state["messages"] + [HumanMessage(content=summary_message)]
60.     # response即为总结内容
61.     response = model.invoke(messages)
62.     # 总结后，仅保留两条消息记录
63.     delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
64.     # delete_messages是最终保留的内容
65.     return {"summary": response.content, "messages": delete_messages}
67. # 定义一个新的graph
68. workflow = StateGraph(State)
70. # 定义节点，会话和总结两个node
71. workflow.add_node("conversation", call_model)
72. workflow.add_node(summarize_conversation)
74. # 定义边，起始节点为conversation，即call_model函数
75. workflow.add_edge(START, "conversation")
77. # 添加条件边
78. workflow.add_conditional_edges(
79.     # 使用conversation，作为起始的node，意味着所有的边都在conversion节点后被访问
80.     "conversation",
81.     # 通过该function，来确认到底执行哪个节点
82.     should_continue,
83. )
85. # summarize_conversation到END是单向边
86. workflow.add_edge("summarize_conversation", END)
88. # 编译，添加参数checkpointer，指定memory
89. app = workflow.compile(checkpointer=memory)
91. # 获取信息
92. def print_update(update):
93.     for k, v in update.items():
94.         for m in v["messages"]:
95.             m.pretty_print()
96.         if "summary" in v:
97.             print(v["summary"])
98. # 定义配置文件，用于后期查询历史
99. config = {"configurable": {"thread_id": "4"}}

复制代码

 调用graph实现对话，当大于6条则总结

1. # 下面三个是问题以及输出
2. input_message = HumanMessage(content="hi! I'm bob")
3. input_message.pretty_print()
4. for event in app.stream({"messages": [input_message]}, config, stream_mode="updates"):
5.     print_update(event)
7. input_message = HumanMessage(content="what's my name?")
8. input_message.pretty_print()
9. for event in app.stream({"messages": [input_message]}, config, stream_mode="updates"):
10.     print_update(event)
12. input_message = HumanMessage(content="i like the celtics!")
13. input_message.pretty_print()
14. for event in app.stream({"messages": [input_message]}, config, stream_mode="updates"):
15.     print_update(event)

复制代码

 输出：

1. [root@Laptop-latitude-7300 chatbot_fastapi]# python test.py
2. ================================ Human Message =================================
4. hi! I'm bob
5. ================================== Ai Message ==================================
7. Hello Bob! How can I assist you today?
8. ================================ Human Message =================================
10. what's my name?
11. ================================== Ai Message ==================================
13. Your name is Bob.
14. ================================ Human Message =================================
16. i like the celtics!
17. ================================== Ai Message ==================================
19. That's great! The Boston Celtics are indeed a fantastic basketball team. They have a rich history and a large fanbase. Are you excited about the upcoming games?
20. ================================ Remove Message ================================
23. ================================ Remove Message ================================
26. ================================ Remove Message ================================
29. ================================ Remove Message ================================
32. In this conversation, the user introduces himself as Bob. He also expresses his liking for the Boston Celtics, a professional basketball team. The assistant acknowledges his interest and asks if he is excited about the upcoming games.

复制代码

 其中pretty_print、print_update是graph格式化的输出，示例如下：
input_message.pretty_print()实现了如下输出

1. ================================ Human Message =================================
3. i like the celtics!

复制代码

print_update(event)输出如下：

1. ================================== Ai Message ==================================
3. Your name is Bob.

复制代码

或
 

1. ================================ Remove Message ================================

复制代码

2、FastAPI、langgraph实现chatbot（非流式）

 基于上述代码，实现非流式输出的问答系统
安装相应的模块

1. pip install fastapi python-dotenv langchain_core langchain langgraph uvicorn langchain_openai

复制代码

 通过fastapi实现接口

1. # main.py
2. from dotenv import load_dotenv
4. load_dotenv(".env")
5. from fastapi import FastAPI, Request, WebSocket
6. from pydantic import BaseModel
7. from json import loads
8. from chat import generate_response, get_chat_history
10. app = FastAPI()
12. # 定义请求和响应模型
13. class ChatRequest(BaseModel):
14.     user_id: str
15.     input: str
17. class ChatResponse(BaseModel):
18.     response: str
20. class ChatHistoryResponse(BaseModel):
21.     history: list
22. # 问答接口
23. @app.get("/chat")
24. async def websocket_ping(query: str, user_id: str):   
25.     response = await generate_response(query, user_id)   
26.     return {"response": response}
28. # 获取历史记录接口
29. @app.get("/chat_history/{user_id}")
30. async def get_history_endpoint(user_id: str):
31.     # 获取聊天历史记录
32.     chat_history = get_chat_history(user_id)
33.    
34.     return {"history": chat_history}
35. if __name__ == "__main__":
36.     import uvicorn
37.     uvicorn.run(app, host="0.0.0.0", port=8000)

复制代码

通过调用generate_response函数，返反响应
 天生响应以及获取汗青记录的函数

1. from langchain_core.messages import HumanMessage, AIMessage
2. from langgraph_test import graphapp
3. from langchain_core.output_parsers import StrOutputParser
4. from langgraph.checkpoint.serde import jsonplus
5. from ignore_socket import ignore_msgpack_default
7. jsonplus._msgpack_default = ignore_msgpack_default
9. # 转换函数, 将数据中的AIMessage，HumanMessage转换为字符串
10. def message_to_dict(message):
11.     if isinstance(message, HumanMessage):
12.         return {
13.             'content': message.content,
14.             'type': 'human',
15.             'additional_kwargs': message.additional_kwargs,
16.             'response_metadata': message.response_metadata,
17.             'id': message.id
18.         }
19.     elif isinstance(message, AIMessage):
20.         return {
21.             'content': message.content,
22.             'type': 'ai',
23.             'additional_kwargs': message.additional_kwargs,
24.             'response_metadata': message.response_metadata,
25.             'id': message.id,
26.             'usage_metadata': message.usage_metadata
27.         }
28.     else:
29.         raise ValueError(f"Unsupported message type: {type(message)}")
31. # 定义生成对话的函数
32. async def generate_response(input_text: str, user_id: str):
34.     # 定义用户配置文件
35.     config = {"configurable": {"thread_id": user_id}}
37.     # 实例化用户输入的消息
38.     input_message = HumanMessage(content=input_text)
40.     # 处理消息
41.     event =  await graphapp.ainvoke({"messages":input_message}, config=config, stream_mode="values")
42.     # 返回信息内容
43.     return event.get("messages")[-1].content
45. # 获取聊天历史记录
46. def get_chat_history(user_id: str):
47.     # 通过config获取相关用户聊天信息
48.     values = graphapp.get_state({"configurable": {"thread_id": user_id}}).values
49.     try:
50.         # 仅获取messages中的content、type数据
51.         values['messages'] = [message_to_dict(message) for message in values['messages']]
52.         values["messages"] = [{k: v for k, v in message.items() if k in ["content", "type"]} for message in values["messages"]]
53.     except:
54.         values["messages"] = []
55.     finally:
56.         return values

复制代码

 使用langgraph实现对话功能

1. from langchain.prompts import PromptTemplate, ChatPromptTemplate
2. from langchain_core.output_parsers import StrOutputParser
3. from typing import Literal
4. from langchain_core.messages import HumanMessage
5. from langchain_openai import AzureChatOpenAI
6. from langchain_core.messages import SystemMessage, RemoveMessage
7. from langgraph.checkpoint.memory import MemorySaver
8. from langgraph.graph import StateGraph, START, END
9. from langgraph.graph.message import add_messages
10. from typing_extensions import TypedDict, Annotated
12. from langchain_core.messages import AnyMessage
13.    
15. memory = MemorySaver()
17. # 我们将使用此模型进行对话和总结
18. llm = AzureChatOpenAI(deployment_name="aicontent-validation")
20. # 定义 Prompt 模板（仅用于输出字符串）
21. prompt_template = PromptTemplate(
22.     input_variables=["query"],
23.     template="You are a helpful assistant. User: {query}. Assistant:"
24. )
26. # 创建 LangChain 的处理链
27. chain = prompt_template | llm | StrOutputParser()
29. # 我们将添加一个 'summary' 属性
30. class State(TypedDict):
31.     messages: Annotated[list[AnyMessage], add_messages]
32.     summary: str
34. # 定义调用模型的逻辑
35. async def call_model(state: State):
36.     # 如果存在摘要，我们会将其作为系统消息添加
37.     summary = state.get("summary", "")
38.     if summary:
39.         system_message = f"Summary of conversation earlier: {summary}"
40.         messages = [SystemMessage(content=system_message)] + state["messages"]
41.     else:
42.         messages = state["messages"]
43.     # 两种调用方式，参数支持{"input":messages}、messages
44.     # response = chain.invoke({"input":messages})
45.     response =await chain.ainvoke(messages)
47.     # 我们返回一个列表，因为它将被添加到现有列表中
48.     return {"messages": [response]}
50. # 现在，我们定义用于确定是结束还是总结对话的逻辑
51. async def should_continue(state: State) -> Literal["summarize_conversation", END]:
52.     """Return the next node to execute."""
53.     messages = state["messages"]
54.     # 如果消息超过 6 条，则我们汇总对话
55.     if len(messages) > 1:
56.         return "summarize_conversation"
57.     # 否则我们就可以结束
58.     return END
61. async def summarize_conversation(state: State):
62.     # 首先，获取总结内容
63.     summary = state.get("summary", "")
64.     if summary:
65.         # 如果 summary 已经存在，我们使用不同的系统提示符总结它
66.         summary_message = (
67.             f"This is summary of the conversation to date: {summary}\n\n"
68.             "Extend the summary by taking into account the new messages above:"
69.         )
70.     else:
71.         summary_message = "Create a summary of the conversation above:"
72.     # 历史信息和总结信息
73.     messages = state["messages"] + [HumanMessage(content=summary_message)]
74.     # 获取总结信息
75.     response = await chain.ainvoke(messages)
76.     # 我们现在需要删除我们不想再显示的消息，将删除除最后两条消息之外的所有消息
77.     delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
78.     return {"summary": response, "messages": delete_messages}
80. # 定义新的graph
81. workflow = StateGraph(State)
83. # 定义 conversation 节点和 summarize 节点
84. workflow.add_node("conversation", call_model)
85. workflow.add_node(summarize_conversation)
87. # 将入口点设置为对话
88. workflow.add_edge(START, "conversation")
90. # 现在，我们添加一条条件边
91. workflow.add_conditional_edges(
92.     # 首先，我们定义起始节点。我们使用 'conversation'。
93.     # 这意味着这些是调用 'conversation' 节点后采用的边。
94.     "conversation",
95.     # 接下来，我们传入一个函数，该函数将确定接下来调用哪个节点。
96.     should_continue,
97. )
99. # 我们现在添加一条从 'summarize_conversation' 到 END 的法线边。
100. # 这意味着在调用 'summarize_conversation' 之后，我们结束。
101. workflow.add_edge("summarize_conversation", END)
103. # 最后，我们编译它！
104. graphapp = workflow.compile(checkpointer=memory)

复制代码

测试功能
启动服务：uvicorn main:app --reload --host 0.0.0.0 --port 8000
浏览器访问：http://172.26.20.199:8000/docs

 
 打开chat接口的折叠，点击try it out，在输入框输入信息，点击下面的执行，稍微等待，如下：

再次执行输入：what is my name，用来测试是否具备影象功能

 
 测试总结功能：使用上述定义的123

在summary中存在总结（择要），成功 
3、FastAPI、langgraph实现chatbot（流式）

基于上述代码，实现基于FastAPI、langchain、langgraph的流式传输，
FastAPI官方文档：WebSocket - FastAPI 中文
 主要鉴戒官方文档，在此基础上实现了流式的传输
安装相应的模块

1. pip install fastapi python-dotenv langchain_core langchain langgraph uvicorn websockets langchain_openai

复制代码

 1、计划html文件，实现websocket请求，基于FastAPI文档魔改

1. // websocket_test.html
3. <!DOCTYPE html>
4. <html lang="en">
6. <head>
7. <meta charset="UTF-8">
8. <meta name="viewport" content="width=device-width, initial-scale=1.0">
9. <title>pc</title>
11. <body>
12. <div id="rowAiChatbot">
13. </div>
15.     <script>
16.         var ws = new WebSocket('ws://localhost:8000/chat');
17.         ws.onopen = function () {
18.             console.log('ws onopen');
19.             ws.send('{"query":"my name is jack", "user_id":"123"}');
20.         };
21.         ws.onmessage = function (e) {
22.             console.log('ws onmessage');
23.             console.log('from server: ' , e.data);
24.         };
25. </script>
26. </body>
27. </html>

复制代码

2、计划FastAPI主程序，实现接收参数并执行

1. # main.py
2. from dotenv import load_dotenv
4. load_dotenv(".env")
5. from fastapi import FastAPI, WebSocket
6. from json import loads
7. from chat import generate_response, get_chat_history
9. app = FastAPI()
11. @app.websocket("/chat")
12. async def websocket_ping(websocket: WebSocket):
13.     await websocket.accept()
14.    
15.     try:
16.         while True:
17.             # 处理消息等逻辑
18.             data = await websocket.receive_text()
19.             jsondata = loads(data)
20.             # 生成响应
21.             await generate_response(jsondata.get("query"), jsondata.get("user_id"), websocket)

复制代码

3、实现调用graph，实现流式输出

鉴戒langgraph文档：LangGraph：Stream events from the final node

1. # chat.py
2. from langchain_core.messages import HumanMessage, AIMessage
3. from langgraph_test import graphapp
6. # 定义生成对话的函数
7. async def generate_response(input_text: str, user_id: str, websocket: object):
9.     # 定义用户配置文件
10.     config = {"configurable": {"thread_id": user_id}}
12.     # 实例化用户输入的消息
13.     input_message = HumanMessage(content=input_text)
15.     # 异步流处理消息
16.     async for msg, metadata in graphapp.astream({"messages":input_message}, config, stream_mode="messages"):
18.         # 仅消息存在且节点为conversation的才返回给前端
19.         if msg.content and metadata.get("langgraph_node") == 'conversation' and metadata.get("ls_provider"):
20.    
21.             # print(msg.content, end=" ")
22.             # 使用websocket实现流式传输到前端
23.             await websocket.send_text(msg.content)

复制代码

4、langgraph实现总结、记录

1. from langchain.prompts import PromptTemplate
2. from langchain_core.output_parsers import StrOutputParser
3. from typing import Literal
4. from langchain_core.messages import HumanMessage, AIMessage
5. from langchain_openai import AzureChatOpenAI
6. from langchain_core.messages import SystemMessage, RemoveMessage
7. from langgraph.checkpoint.memory import MemorySaver
8. from langgraph.graph import MessagesState, StateGraph, START, END
10. memory = MemorySaver()
12. # 我们将使用此模型进行对话和总结
13. llm = AzureChatOpenAI(deployment_name="aicontent-validation")
15. # 定义 Prompt 模板（仅用于输出字符串）
16. prompt_template = PromptTemplate(
17.     input_variables=["query"],
18.     template="You are a helpful assistant. User: {query}. Assistant:"
19. )
21. # 创建 LangChain 的处理链
22. chain = prompt_template | llm | StrOutputParser()
24. # 我们将添加一个 'summary' 属性（除了MessagesState已经有 'messages' 键之外）
25. class State(MessagesState):
26.     summary: str
28. # 定义调用模型的逻辑
29. async def call_model(state: State):
31.     # 如果存在摘要，我们会将其作为系统消息添加
32.     summary = state.get("summary", "")
33.    
34.     if summary:
35.         system_message = f"Summary of conversation earlier: {summary}"
36.         messages = [SystemMessage(content=system_message)] + state["messages"]
37.     else:
38.         messages = state["messages"]
39.     # 两种调用方式，参数支持{"input":messages}、messages
40.     # response = chain.invoke({"input":messages})
41.     response =await chain.ainvoke(messages)
42.     # print(response)
43.     # 用于验证输出一致，此处执行与generate_response位置输出时间基本一致
44.     # async for msg in chain.astream(messages):
45.     #     print(msg)
46.     # 我们返回一个列表，因为它将被添加到现有列表中
47.     return {"messages": [AIMessage(content=response)]}
49. # 现在，我们定义用于确定是结束还是总结对话的逻辑
50. async def should_continue(state: State) -> Literal["summarize_conversation", END]:
51.     """Return the next node to execute."""
52.     messages = state["messages"]
53.     # 如果消息超过 6 条，则我们汇总对话
54.     if len(messages) > 3:
55.         return "summarize_conversation"
56.     # 否则我们就可以结束
57.     return END
59. async def summarize_conversation(state: State):
60.     # 首先，获取总结内容
61.     summary = state.get("summary", "")
62.     if summary:
63.         # 如果 summary 已经存在，我们使用不同的系统提示符总结它
64.         summary_message = (
65.             f"This is summary of the conversation to date: {summary}\n\n"
66.             "Extend the summary by taking into account the new messages above:"
67.         )
68.     else:
69.         summary_message = "Create a summary of the conversation above:"
70.     # 历史信息和总结信息
71.     messages = state["messages"] + [HumanMessage(content=summary_message)]
72.     # 获取总结信息
73.     response = await chain.ainvoke(messages)
74.     # 我们现在需要删除我们不想再显示的消息，将删除除最后两条消息之外的所有消息
75.     delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
76.     return {"summary": response, "messages": delete_messages}
78. # 定义新的graph
79. workflow = StateGraph(State)
81. # 定义 conversation 节点和 summarize 节点
82. workflow.add_node("conversation", call_model)
83. workflow.add_node(summarize_conversation)
85. # 将入口点设置为对话
86. workflow.add_edge(START, "conversation")
88. # 现在，我们添加一条条件边
89. workflow.add_conditional_edges(
90.     # 首先，我们定义起始节点。我们使用 'conversation'。
91.     # 这意味着这些是调用 'conversation' 节点后采用的边。
92.     "conversation",
93.     # 接下来，我们传入一个函数，该函数将确定接下来调用哪个节点。
94.     should_continue,
95. )
97. # 我们现在添加一条从 'summarize_conversation' 到 END 的法线边。
98. # 这意味着在调用 'summarize_conversation' 之后，我们结束。
99. workflow.add_edge("summarize_conversation", END)
101. # 最后，我们编译它！
102. graphapp = workflow.compile(checkpointer=memory)

复制代码

5、测试流式传输

使用浏览器打开html文件，效果如下：

6、获取汗青记录

 请求接口

2. @app.get("/chat_history/{user_id}")
3. async def get_history_endpoint(user_id: str):
4.     # 获取聊天历史记录
5.     chat_history = get_chat_history(user_id)
6.    
7.     return {"history": chat_history}

复制代码

 获取逻辑方法，定义一个函数进行数据转换

1. # chat.py
3. # 转换函数, 将数据中的AIMessage，HumanMessage转换为字符串
4. def message_to_dict(message):
5.     if isinstance(message, HumanMessage):
6.         return {
7.             'content': message.content,
8.             'type': 'human',
9.             'additional_kwargs': message.additional_kwargs,
10.             'response_metadata': message.response_metadata,
11.             'id': message.id
12.         }
13.     elif isinstance(message, AIMessage):
14.         return {
15.             'content': message.content,
16.             'type': 'ai',
17.             'additional_kwargs': message.additional_kwargs,
18.             'response_metadata': message.response_metadata,
19.             'id': message.id,
20.             'usage_metadata': message.usage_metadata
21.         }
22.     else:
23.         raise ValueError(f"Unsupported message type: {type(message)}")
26. # 获取聊天历史记录
27. def get_chat_history(user_id: str):
28.     # 通过config获取相关用户聊天信息
29.     values = graphapp.get_state({"configurable": {"thread_id": user_id}}).values
30.     try:
31.         # 仅获取messages中的content、type数据
32.         values['messages'] = [message_to_dict(message) for message in values['messages']]
33.         values["messages"] = [{k: v for k, v in message.items() if k in ["content", "type"]} for message in values["messages"]]
34.     except:
35.         values["messages"] = []
36.     finally:
37.         return values

复制代码

浏览器测试
网址：http://172.25.23.143:8000/docs#

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。