from langchain.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
system_msg = SystemMessage("You are a helpful assistant. Answer in one sentence.") human_msg = HumanMessage( content="What's the weather in Paris?", name="alice", id="msg_human_001", ) ai_msg = AIMessage( content=[], tool_calls=[{ "name": "get_weather", "args": {"location": "Paris"}, "id": "call_abc123", }], ) tool_msg = ToolMessage( content="Sunny, 22C", tool_call_id="call_abc123", name="get_weather", )
for m in (system_msg, human_msg, ai_msg, tool_msg): print(type(m).__name__, "| type =", m.type, "| id =", m.id) print(" content :", repr(m.content)) print(" content_blocks :", m.content_blocks)
---- SystemMessage ---- type : system id : None content : 'You are a helpful assistant. Answer in one sentence.' content_blocks : [{'type': 'text', 'text': 'You are a helpful assistant. Answer in one sentence.'}]
---- HumanMessage ---- type : human id : msg_human_001 content : "What's the weather in Paris?" content_blocks : [{'type': 'text', 'text': "What's the weather in Paris?"}]
---- final AIMessage ---- content : "The weather in Paris is currently **sunny with a temperature of 22°C** (about 72°F). It's a lovely day there!" usage : {'input_tokens': 339, 'output_tokens': 35, 'total_tokens': 374, ...} tool_calls: []
DeepSeek 生成的调用 ID 长这样:call_00_m5sEphpzABSo8NpneDTG4455。你把它原样抄进 ToolMessage.tool_call_id,配对就成立。手工构造时常用的 call_123 这种短 ID 只是示例写法,真实 API 返回的字符串不要改。
故意把 ID 写错会怎样,我试了:
1 2 3 4 5 6 7 8 9 10 11
bad_messages = [ HumanMessage("What's the weather in Paris?"), ai_message, ToolMessage(content="Sunny, 22C", tool_call_id="call_wrong_id", name="get_weather"), ] try: r = model_with_tools.invoke(bad_messages) print("no error, content:", r.content) except Exception as e: print(type(e).__name__) print(str(e)[:800])
真实报错:
1 2
OpenAIInvalidRequestError Error code: 400 - {'error': {'message': "An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id', The following tool_call_ids did not have response messages: call_wrong_id (request_id: 39d6737c-8128-437c-ba84-41f7b65b2b7f)", 'type': 'invalid_request_error', 'param': None, 'code': 'invalid_request_error'}}
报错信息有点绕:它说的是 call_wrong_id 没有对应的响应消息。因为模型看到的历史里,call_wrong_id 这个 ID 从来没被调用过,你却在回应它;而真正被调用的那个 ID,你又没回。服务端两边都对不上,直接 400。
messages = [ SystemMessage("You are a terse assistant. Answer in one short sentence."), HumanMessage("What is the capital of France?"), AIMessage("Paris."), HumanMessage("And of Japan?"), ] response = model.invoke(messages)
response = model.invoke([ {"role": "system", "content": "You are a terse assistant."}, {"role": "user", "content": "Reply with the single word: pong"}, ]) print(type(response).__name__, repr(response.content)) # AIMessage 'pong'
反向转换用 convert_to_messages,它按 role 把字典变成对应的消息对象:
1 2 3 4 5 6 7 8 9 10
from langchain_core.messages import convert_to_messages
raw = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi"}, {"role": "assistant", "content": "Hello! How can I help?"}, {"role": "tool", "content": "Sunny, 22C", "tool_call_id": "call_abc123"}, ] for o in convert_to_messages(raw): print(type(o).__name__, "| type =", o.type, "| tool_call_id =", getattr(o, "tool_call_id", None))
真实输出:
1 2 3 4
SystemMessage | type= system | content= 'You are a helpful assistant.' | tool_call_id= None HumanMessage | type= human | content= 'Hi' | tool_call_id= None AIMessage | type= ai | content= 'Hello! How can I help?' | tool_call_id= None ToolMessage | type= tool | content= 'Sunny, 22C' | tool_call_id= call_abc123
flowchart LR
D["dict: role / content"] -->|convert_to_messages| O[消息对象]
O -->|model_dump| D2["普通 dict, 不含类信息"]
O -->|dumpd| S["带 lc 标记的 dict"]
S -->|load| O2[还原的消息对象]
from langchain.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.messages import trim_messages
history = [ SystemMessage("You are a helpful assistant."), HumanMessage("Hi, my name is Bob."), AIMessage("Nice to meet you, Bob."), HumanMessage("Write a poem about cats."), AIMessage("Cats sit on walls..."), HumanMessage("Now do the same for dogs."), AIMessage("Dogs run in parks..."), HumanMessage("What is my name?"), ]
trimmed = trim_messages( history, max_tokens=5, strategy="last", token_counter=len, include_system=True, start_on="human", ) print("original:", len(history), "-> trimmed:", len(trimmed)) for m in trimmed: print(" ", type(m).__name__, repr(m.content)[:60])
真实输出:
1 2 3 4 5 6 7 8 9 10 11 12 13
original length: 8 trimmed length: 4 (token_counter=len, max_tokens=5, strategy=last) SystemMessage 'You are a helpful assistant.' HumanMessage 'Now do the same for dogs.' AIMessage 'Dogs run in parks...' HumanMessage 'What is my name?'
include_system=False -> length: 5 HumanMessage 'Write a poem about cats.' AIMessage 'Cats sit on walls...' HumanMessage 'Now do the same for dogs.' AIMessage 'Dogs run in parks...' HumanMessage 'What is my name?'
from langchain.messages import RemoveMessage from langgraph.graph.message import REMOVE_ALL_MESSAGES from langchain.agents.middleware import before_model
@before_model deftrim_messages(state, runtime): """Keep only the last few messages to fit context window.""" messages = state["messages"] iflen(messages) <= 3: returnNone first_msg = messages[0] recent_messages = messages[-3:] iflen(messages) % 2 == 0else messages[-4:] return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), first_msg, *recent_messages]}
删除会丢信息,文档说得很直白:裁剪掉的消息里的信息就没了。上面那段 include_system=False 的输出里,Hi, my name is Bob. 被裁掉了,如果后面接着问”我叫什么”,模型只能回答不知道。信息必须留住,就换成 SummarizationMiddleware,让模型把早期历史压成摘要。
裁剪还有一条硬约束:结果历史必须合法。有的 provider 要求历史以 user 消息开头;带工具调用的 assistant 消息后面必须跟对应的 tool 结果。裁剪的边界要是正好切在这对中间,下一个请求就是 400。
直接调模型,还是交给 agent
前面所有例子都是直接调模型,工具执行、消息拼装全靠自己。换成 create_agent 会怎样:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent( model=model, tools=[get_weather], system_prompt="You are a helpful assistant.", checkpointer=InMemorySaver(), ) config = {"configurable": {"thread_id": "demo-1"}} result = agent.invoke( {"messages": [{"role": "user", "content": "What's the weather in Paris?"}]}, config, ) for m in result["messages"]: print(type(m).__name__, "| content =", repr(m.content)[:60], "| tool_call_id =", getattr(m, "tool_call_id", None))
真实输出:
1 2 3 4 5 6 7 8 9 10 11 12
---- direct model call ---- returned type : AIMessage content : "I'll check the current weather in Paris for you." tool_calls : [{'name': 'get_weather', 'args': {'location': 'Paris'}, 'id': 'call_00_QWOud95DNoG98VytHC2i0841', 'type': 'tool_call'}]
---- through an agent ---- state keys : ['messages'] message count : 4 HumanMessage | content= "What's the weather in Paris?" | tool_calls= None | tool_call_id= None AIMessage | content= "I'll check the current weather in Paris for you." | tool_calls= [{...}] | tool_call_id= None ToolMessage | content= 'Sunny, 22C' | tool_calls= None | tool_call_id= call_00_NRo1kEWP2HT5qlQDMdDm8013 AIMessage | content= "The weather in Paris is currently sunny with a temperature of 22°C (72°F)." | tool_calls= [] | tool_call_id= None
---- follow-up turn on the same thread ---- message count : 6 last content : "I'm sorry, but I can only retrieve the **current** weather - I don't have access to forecast data for tomorrow. ..."
两条路的差别在于谁持有历史。直接调模型,历史是你手里的一个列表,随时能改、能存、能裁剪。用 agent,历史在图的 state 里,append-only,改动要走中间件和 RemoveMessage。要做精细控制,比如手工插入伪造的 AIMessage、按业务规则改写历史,直接调模型更顺手。要做多轮工具循环,agent 省掉的配对和循环代码不少。