开启人工智能大模型应用开发新篇章：运用LangChain构建聊天机器人
在大型模型应用的开发进程里，状态管理以及对话追踪是不容忽视的关键能力，特别是在需要保存上下文、重放对话或者进行异步处理的时候，其重要性尤为凸显。
当下，我们来演示怎样利用LangChain、OpenAI的GPT模型以及PostgreSQL搭建一个具备消息存储功能的聊天机器人。
技术栈
-
LangChain（LangGraph组件）
-
OpenAI GPT-4o-mini（或其他兼容的模型）
-
PostgreSQL（用于状态和消息的持久化存储）
-
Python 3.10及以上版本
目标功能
我们将要达成的能力如下：
-
构建一个简洁却可扩展的对话工作流程
-
利用PostgreSQL存储多轮对话的状态信息
-
支持依据thread_id恢复上下文内容
-
模型自动对历史消息进行裁剪，确保Token数量不超出限制
-
模拟一个具有“卡通风格”语气的AI角色来进行回复
目录结构
.
├── llm_env.py # 初始化大模型
├── chatbot_base_on_postgres_demo.py # 主对话逻辑
初始化LLM环境（llm_env.py）
我们运用langchain.chat_models.init_chat_model来加载GPT-4o-mini：
from langchain.chat_models import init_chat_model
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）
导入依赖
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_core.messages import HumanMessage, BaseMessage, trim_messages
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
状态定义与裁剪器（Token管理）
我们定义对话状态，并且借助LangChain的trim_messages来限制最大的token数量，防止模型请求出现失败情况：
trimmer = trim_messages(
max_tokens=65,
strategy="last",
token_counter=model,
include_system=True,
allow_partial=False,
start_on="human",
)
Prompt模板
prompt_template = ChatPromptTemplate.from_messages(
[
("system", "你说话得像个卡通角色似的。尽你所能按照{language}语言回答所有问题。"),
MessagesPlaceholder(variable_name="messages"),
]
)
定义状态结构
class State(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
language: str
工作流定义
我们使用LangGraph定义一个极为简单的图：
workflow = StateGraph(state_schema=State)
def call_model(state: State):
trimmed = trimmer.invoke(state["messages"])
prompt = prompt_template.invoke({
"messages": trimmed,
"language": state["language"]
})
response = model.invoke(prompt)
return {"messages": [response]}
workflow.add_edge(START, "call_model")
workflow.add_node("call_model", call_model)
接入PostgreSQL存储状态
DB_URI = "postgresql://postgres:123456@localhost:5432/langchaindemo?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
app = workflow.compile(checkpointer=checkpointer)
LangGraph的PostgresSaver会在数据库中为每一个thread_id维护一份完整的状态记录（状态压缩加上消息追踪），非常适合用于审计和调试。
启动聊天循环
input_thread_id = input("输入 thread_id: ")
config = {"configurable": {"thread_id": input_thread_id}}
while True:
query = input("你: ")
if query.strip().lower() == "exit":
break
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages, "language": "中文"}, config)
for message in output["messages"]:
print(f"{message.type}: {message.content}")
你可以多次使用同一个thread_id调用app.invoke，以此来恢复之前的上下文继续进行对话。
效果展示
输入 thread_id: abc123
你: 今天天气怎么样？
ai: 哇哦，阳光灿烂的好天气最适合出去玩耍啦！☀️
你: 你是谁呀？
ai: 我是你的超萌聊天小帮手，随时准备开启冒险之旅！✨
总结
这仅仅是LangGraph的一个入门示例，但其潜力远不止于此。你可以：
-
添加更多的节点（比如知识检索、函数调用、代码执行等）
-
与前端（像Gradio / Streamlit）相结合打造完整的应用
-
利用数据库来管理多用户的会话历史
