短期记忆

from langchain_core.messages.utils import trim_messages, count_tokens_approximately
from langchain_core.messages import BaseMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain_core.runnables import RunnableConfig

def pre_model_hook(state) -> dict[str, list[BaseMessage]]:
    """
    此函数将在每次调用LLM之前被调用，以准备LLM的消息。
    """
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",
        token_counter=count_tokens_approximately,
        max_tokens=384,
        start_on="human",
        end_on=("human", "tool"),
    )
    return {"llm_input_messages": trimmed_messages}


checkpointer = InMemorySaver()
agent = create_agent(
    "openai:gpt-5-nano",
    tools=[],
    pre_model_hook=pre_model_hook,
    checkpointer=checkpointer,
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()
"""
================================== Ai Message ==================================

您的名字是Bob。您之前告诉过我。
如果您希望我称呼您昵称或使用其他名字，请随时告诉我。
"""

from langchain_core.messages import RemoveMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig


def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        # 删除最早的两条消息
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}


agent = create_agent(
    "openai:gpt-5-nano",
    tools=[],
    prompt="请简洁明了。",
    post_model_hook=delete_messages,
    checkpointer=InMemorySaver(),
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}

for event in agent.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

for event in agent.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

[('human', "hi! I'm bob")]
[('human', "hi! I'm bob"), ('ai', '嗨，Bob！很高兴认识你。今天有什么我可以帮你的吗？我可以回答问题、头脑风暴、起草文本、解释事物或帮助编写代码。')]
[('human', "hi! I'm bob"), ('ai', '嗨，Bob！很高兴认识你。今天有什么我可以帮你的吗？我可以回答问题、头脑风暴、起草文本、解释事物或帮助编写代码。'), ('human', "what's my name?")]
[('human', "hi! I'm bob"), ('ai', '嗨，Bob！很高兴认识你。今天有什么我可以帮你的吗？我可以回答问题、头脑风暴、起草文本、解释事物或帮助编写代码。'), ('human', "what's my name?"), ('ai', '您的名字是Bob。今天有什么我可以帮您的吗，Bob？')]
[('human', "what's my name?"), ('ai', '您的名字是Bob。今天有什么我可以帮您的吗，Bob？')]

from typing import Annotated
from langchain.agents import create_agent, AgentState
from langchain.agents.tool_node import InjectedState

class CustomState(AgentState):
    user_id: str

def get_user_info(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """查找用户信息。"""
    user_id = state["user_id"]
    return "用户是John Smith" if user_id == "user_123" else "未知用户"

agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_user_info],
    state_schema=CustomState,
)

result = agent.invoke({
    "messages": "查找用户信息",
    "user_id": "user_123"
})
print(result["messages"][-1].content)
# > 用户是John Smith。

from langchain_core.messages import AnyMessage
from langchain.agents import create_agent, AgentState
from langgraph.runtime import get_runtime
from typing import TypedDict


class CustomContext(TypedDict):
    user_name: str


def get_weather(city: str) -> str:
    """获取城市的天气。"""
    return f"{city}的天气总是晴朗的！"


def prompt(state: AgentState) -> list[AnyMessage]:
    user_name = get_runtime(CustomContext).context["user_name"]
    system_msg = f"您是一个有用的助手。请称呼用户为 {user_name}。"
    return [{"role": "system", "content": system_msg}] + state["messages"]


agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_weather],
    prompt=prompt,
    context_schema=CustomContext,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "旧金山的天气如何？"}]},
    context=CustomContext(user_name="John Smith"),
)
for msg in result["messages"]:
    msg.pretty_print()

from langchain_core.messages.utils import trim_messages, count_tokens_approximately
from langchain_core.messages import BaseMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState


def pre_model_hook(state: AgentState) -> dict[str, list[BaseMessage]]:
    """
    此函数将在每次调用LLM之前被调用，以准备LLM的消息。
    """
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",
        token_counter=count_tokens_approximately,
        max_tokens=384,
        start_on="human",
        end_on=("human", "tool"),
    )
    return {"llm_input_messages": trimmed_messages}

agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[],
    pre_model_hook=pre_model_hook,
    checkpointer=InMemorySaver(),
)

result = agent.invoke({"messages": "hi, my name is bob"}, {"configurable": {"thread_id": "1"}})
print(result["messages"][-1].content)

from langchain.agents import create_agent, AgentState

STOP_WORDS = ["密码", "秘密"]

def validate_response(state: AgentState) -> dict[str, list[BaseMessage]]:
    """确认响应内容不包含停用词列表中的任何内容。"""
    last_message = state["messages"][-1]
    if any(word in last_message.content for word in STOP_WORDS):
        return {"messages": [RemoveMessage(id=last_message.id)]}
    return {}

agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[],
    post_model_hook=validate_response,
    checkpointer=InMemorySaver(),
)