链能查资料了,但每次调用默认都是独立的:上一轮说过的话,不会自动进入下一轮;要获取实时数据、调一次外部接口,模型自己也做不到。
这两件事都不在模型内部。模型本身是无状态的,每次调用只能看到这一次传入的上下文。
所以,“记忆”和“工具”都由应用层负责:历史由应用保存并重新带回上下文,工具则由应用提供并执行。
#一、模型是无状态的,“记忆”就是把历史重新带回上下文
先看没有记忆时会发生什么:
chat_template = ChatPromptTemplate.from_messages(
[
SystemMessagePromptTemplate.from_template("你是人工智能助手"),
('human', '{text}')
]
)
parser = StrOutputParser()
chain = chat_template | client | parser
chain.invoke({'text': '你好,我是muzi'})
chain.invoke({'text': '你好,我是谁?'})
第二次调用收到的只有 '你好,我是谁?'。第一轮提到的名字没有被带进来,模型不知道“mu zi”是谁。
每一次 invoke 都是一次独立请求:输入进去什么,模型就只能看到什么。
要让它第二轮知道“muzi”是谁,本质上就是把前边的对话历史重新放进这一次请求的上下文。
#二、把历史交给组件托管,并按会话隔离
历史本质上是一组消息。要让模型看到之前的对话,可以在Prompt中给历史留一个位置:
chat_template = ChatPromptTemplate.from_messages(
[
("system", "你是人工智能助手"),
# 作用就是向提示词中插入一段上下文消息
MessagesPlaceholder(variable_name="messages"),
("human", "{input}"),
]
)
MessagesPlaceholder 接收的是一组历史消息。每次调用时,历史放进 history,当前用户输入放进 input。
手动管理时,大致要做三件事:
chat_history.add_user_message(user_input)
response = chain.invoke({
"messages": chat_history.messages,
"input": user_input
})
chat_history.add_ai_message(response)
也就是:先保存用户消息,再调用模型,最后保存模型回复。
每一轮都手写容易遗漏,所以 langChain 提供了 RunnableWithMessageHistory,把读取历史和写回历史这两件事包起来。
def get_session_history(session_id):
return RedisChatMessageHistory(
session_id=session_id,
url="redis://localhost:6379",
ttl=300
)
chatbot_with_his = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
)
这里几个参数各管一件事:
chain:真正执行对话的链;get_session_history:根据对话 ID 找到对应的历史;input_messages_key="input":指出本轮用户输入在哪。
调用时,通过 config 里的 session_id 区分不同的回话:
chatbot_with_his.invoke(
{"input": "我喜欢音乐,你能给我推荐一首轻音乐吗?"},
config={"configurable": {"session_id": "user123"}},
)
RedisChatMessageHistory 默认就是通过 session_id 找到对应的会话历史。不同 session_id 对应不同的对话历史,所以多个用户或多个会话不会串在一起。
历史对话的存储位置,由 get_session_history 决定。放在内存里,进程退出就没了;放在 Redis 这类外部存储里,则可以跨进程保留。
早期 LangChain 常用 ConversationBufferMemory 这类 Memory 组件,现在已经被标记为 deprecated;当前这类 LCEL 场景更适合使用消息历史配合 RunnableWithMessageHistory。
所以“记忆”,本质上不是模型自己记住了什么,而是应用根据 session_id 保存历史,并在一下轮调用时重新把它放进上下文。
#三、工具的几个核心要素
工具是 Agent 或模型与外部能力交互的接口。一个工具通常包含:
- 名称:模型识别这个工具的标识;
- 描述:用来说明这个工具什么时候该用;
- 输入schema:定义需要哪些参数、参数是什么类型;
- 执行函数:真正执行外部逻辑的代码;
- 返回结果:工具执行后返回给模型或者应用的数据。
其中最容易被忽略的是描述。模型通常会根据工具名称、描述和参数结构来判断这个问题该不该交给这个工具,它并不会读取函数体来理解工具用途。
定义工具最简单的方式,使用 @tool:
from langchain_core.tools import Tool
@tool
def get_date():
""" 获取今天的具体日期 """
return datetime.date.today().strftime("%Y-%m-%d")
@tool
def open_browser(url, browser_name=None):
""" 获取浏览器,打开网站 """
使用 @tool 时,如果没有额外指定描述,函数的 doctoring 会作为工具描述,所以工具用途要写清楚。
也可以直接构造 Tool:
get_time_tool = Tool(
name="get_current_time",
func=get_current_time,
description="当你想知道现在的时间时非常有用。"
)
两种方式最终得到的都是 Tool 对象,后面都可以绑定给模型或交给 Agent 使用。
#四、bind_tools 只负责选,执行由应用负责
把工具绑定给模型后,就可以让模型判断什么时候需要调用工具:
tool_llm = model.bind_tools([get_date, open_browser])
resp = tool_llm.invoke("今天是几月几号?")
如果模型决定要调用工具,resp.tool_calls 里会包含它选中的工具名和参数:
[
{
"name": "get_date",
"args": {},
...
}
]
模型只负责判断“该用哪个工具、参数填什么”,并不会真正执行工具。
要拿到工具结果,还需要应用根据 tool_calls 自己执行:
all_tools = {
"get_date": get_date,
"open_browser": open_browser
}
if resp.tool_calls:
for tool_call in resp.tool_calls:
selected_tool = all_tools.get(tool_call["name"])
result = selected_tool.invoke(tool_call["args"])
所以,绑定不等于执行。 bind_tools 只是把工具的名称、描述和参数结构告诉模型,让它具备提出工具调用请求的能力。
一次完整的工具调用往往不止一步。如果任务需要连续调用多个工具,这个循环就会越来越长。
这些步骤可以手写实现,但是工具多了、步骤变长,控制逻辑会迅速变复杂。下一步就需要 Agent 来接管这套循环。
#五、Agent 把工具调用循环封装起来
前边手写工具调用时,需要自己处理:
模型选择工具
→ 应用执行工具
→ 把结果交回模型
→ 模型继续判断
create_agent 把这套循环封装起来:
agent = create_agent(
model,
tools=[get_date, open_browser],
)
result = agent.invoke({"messages":[{"role":"user","content":"帮我打开淘宝"}]})
一次 invoke 里,Agent 会根据模型返回的 tool_calls 自动执行工具,再把工具结果交给模型;如果模型还需要调用别的工具,就继续循环,直到得到最终回答。
所以 Agent 做的事情,本质上是:
判断 → 调用工具 → 拿到结果 → 继续判断 → 最终回答
create_agent 接收的 model 和 tools,就是前面已经见过的聊天模型和工具列表。
如果还需要保留多轮会话状态,可以给 Agent 加上 checkpointer:
agent = create_agent(
model=llm,
tools=tools,
system_prompt=system_prompt,
checkpointer=memory
)
调用时用 thread_id 区分不同会话:
result = agent.invoke(
{"messages": [{"role": "user", "content": query}]},
config={"configurable": {"thread_id": "user_1"}}
)
checkpointer 负责保存这个会话线程的状态,thread_id 则用来找到对应的历史。它表示的是会话,不是用户本身;同一个用户也可以有多个 thread。
但不是所有场景都需要 Agent。
如果执行路径本来就是固定的,例如:
问题 → 生成 SQL → 执行 SQL → 根据结果回答
那就直接写成一条 Chain:
sql_make_chain = create_sql_query_chain(client, db) | SQLCleaner()
execute_sql_tools = QuerySQLDatabaseTool(db=db)
chain = (RunnablePassthrough.assign(query=sql_make_chain)
.assign(result=itemgetter('query') | execute_sql_tools)
| answer_prompt | client | StrOutputParser())
这类流程里,每一步都确定,不需要模型再额外判断“这次要不要调用工具”。
路径固定,用 Chain;路径需要模型动态判断,用 Agent。
#六、历史长了怎么办
历史只增不减,迟早会碰到上下文窗口上限。全部保留不现实,直接截掉又容易丢失前面的关键信息。
SummarizationMiddleware 处理的就是这一步:历史接近上限时,把较早的历史消息压缩成摘要,同时保留最近的消息。**
memory = InMemorySaver()
agent = create_agent(
model=llm,
tools=[],
checkpointer=memory,
middleware=[
SummarizationMiddleware(
model=llm,
trigger=("tokens", 4000), # 达到 4000 token 时触发摘要
keep=("messages", 10), # 保留最近 10 条消息
summary_prompt="请将以下对话历史进行简洁的摘要,保留关键信息: {messages}"
),
],
)
trigger:什么时候触发摘要;keep:摘要后保留多少最近消息;summary_prompt:是这次摘要用的提示词。
整个过程可以理解成:完整历史越来越长,达到阈值,就消息做摘要,最近的消息原样保留,继续参与后续对话。这样模型看到的不是无限增长的完整历史,而是历史摘要加最近的几轮原始消息。
在 Agent 里,checkpointer 负责保存会话状态,SummarizationMiddleware 负责在上下文过长时压缩历史。两者解决的是不同问题。
#七、总结
- 模型本身无状态。历史由应用在每轮调用时重新带回上下文;历史可以存在内存、Redis等不同存储中,上层调用方式基本不变。
- 工具绑定不等于工具执行。模型根据工具名称、描述和参数结构决定是否调用以及参数怎么填,
bind_tools只产生调用请求,真正的执行仍由应用负责。 - “历史上下文存在哪”和“历史上下文太长怎么办”是两个独立问题,前者由消息历史或者
checkpointer负责,后者由摘要中间件处理。
到这里,一条能记住对话、能调用外部工具的链就搭好了。
但它目前还只是应用内部的一段能力,还没有真正对外提供服务。