AI大模型应用开发入门-LangChain开发聊天机器人ChatBot

在大模型应用开发中，状态管理 和 对话追踪 是不可忽视的重要能力，尤其在需要保存上下文、重放对话或进行异步处理时尤为关键。
今天我们来演示如何用 LangChain + OpenAI 的 GPT 模型 + PostgreSQL 搭建一个 具备消息存储能力 的聊天机器人。
技术栈

• LangChain（LangGraph 模块）
  
• OpenAI GPT-4o-mini（或任意兼容模型）
  
• PostgreSQL（用于状态和消息持久化）
  
• Python 3.10+
  

目标功能

我们将实现如下能力：

• 构建一个简单但可扩展的对话工作流
  
• 用 PostgreSQL 存储多轮对话状态
  
• 支持根据 thread_id 恢复上下文
  
• 模型自动裁剪历史消息，保证 Token 数不超限
  
• 模拟一个“卡通语气”的 AI 角色进行回复
  

目录结构

1. .
2. ├── llm_env.py                      # 初始化大模型
3. ├── chatbot_base_on_postgres_demo.py  # 主对话逻辑

复制代码

初始化 LLM 环境（llm_env.py）

我们用 langchain.chat_models.init_chat_model 来加载 GPT-4o-mini：

1. from langchain.chat_models import init_chat_model
3. llm = init_chat_model("gpt-4o-mini", model_provider="openai")

复制代码

你可以替换为任意你有权限使用的模型，比如 gpt-3.5-turbo, gpt-4, claude, mistral，甚至本地模型如 Ollama。

定义对话流程（chatbot_base_on_postgres_demo.py）

导入依赖

1. from langgraph.graph import START, MessagesState, StateGraph
2. from langgraph.graph.message import add_messages
3. from langgraph.checkpoint.postgres import PostgresSaver
4. from langchain_core.messages import HumanMessage, BaseMessage, trim_messages
5. from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

复制代码

状态定义与裁剪器（Token 管理）

我们定义对话状态，并用 LangChain 的 trim_messages 限制最大 token 数量，避免模型请求失败：

1. trimmer = trim_messages(
2.     max_tokens=65,
3.     strategy="last",
4.     token_counter=model,
5.     include_system=True,
6.     allow_partial=False,
7.     start_on="human",
8. )

复制代码

Prompt 模板

1. prompt_template = ChatPromptTemplate.from_messages(
2.     [
3.         ("system", "你说话像个卡通人物。尽你所能按照语言{language}回答所有问题。"),
4.         MessagesPlaceholder(variable_name="messages"),
5.     ]
6. )

复制代码

定义状态结构

1. class State(TypedDict):
2.     messages: Annotated[list[BaseMessage], add_messages]
3.     language: str

复制代码

工作流定义

我们用 LangGraph 定义一个非常简单的图：

1. workflow = StateGraph(state_schema=State)
3. def call_model(state: State):
4.     trimmed = trimmer.invoke(state["messages"])
5.     prompt = prompt_template.invoke({
6.         "messages": trimmed,
7.         "language": state["language"]
8.     })
9.     response = model.invoke(prompt)
10.     return {"messages": [response]}
12. workflow.add_edge(START, "call_model")
13. workflow.add_node("call_model", call_model)

复制代码

接入 PostgreSQL 存储状态

1. DB_URI = "postgresql://postgres:123456@localhost:5432/langchaindemo?sslmode=disable"
2. with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
3.     checkpointer.setup()
5.     app = workflow.compile(checkpointer=checkpointer)

复制代码

LangGraph 的 PostgresSaver 会在数据库中为每个 thread_id 维护一份完整的状态记录（状态压缩 + 消息追踪），非常适合审计和调试。
启动聊天循环

1. input_thread_id = input("输入 thread_id: ")
2. config = {"configurable": {"thread_id": input_thread_id}}
4. while True:
5.     query = input("你: ")
6.     if query.strip().lower() == "exit":
7.         break
8.     input_messages = [HumanMessage(query)]
9.     output = app.invoke({"messages": input_messages, "language": "中文"}, config)
10.     for message in output["messages"]:
11.         print(f"{message.type}: {message.content}")

复制代码

你可以使用同一个 thread_id 多次调用 app.invoke，来恢复之前的上下文继续对话。
效果展示

1. 输入 thread_id: abc123
2. 你: 今天天气怎么样？
3. ai: 哇哦，阳光明媚的天气最适合出去玩啦！☀️
4. 你: 你是谁呀？
5. ai: 我是你的超级可爱聊天小助手，随时准备出发冒险！✨

复制代码

总结

这只是 LangGraph 的一个入门示例，但它的潜力远不止如此。你可以：

• 添加更多节点（如知识检索、函数调用、代码执行等）
  
• 与前端（如 Gradio / Streamlit）结合打造完整应用
  
• 利用数据库管理多用户会话历史
  

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

谢谢楼主提供！

不错，里面软件多更新就更好了

收藏一下   不知道什么时候能用到