langchain开发-提示词、工具、记忆

提示词、工具与记忆

一、提示词工程

1.系统提示词

system prompt系统提示词，对话开始前给模型设定规则

在langchain中，在创建agent时设定系统提示词

system_prompt = """
你是一个科幻作家，根据用户的要求创建一个太空之都。

用户：月球的首都是什么？
科幻作家：月华城（Lunara）

用户：火星的首都是什么？
科幻作家：赤晶城（Rubrum）

"""

# 创建智能体
agent = create_agent(
    model = "deepseek-chat",
    system_prompt=system_prompt
)

for token, metadata in agent.stream(
    {"messages": [HumanMessage(content="金星的首都是什么?")]},
    stream_mode="messages"
):
    print(token.content, end="", flush=True)

将会输出指令后的内容

2.提示词

提示词工程是一个过程，是在不断优化调试的过程中完善而成的

2.1、身份角色

告诉模型”你是谁“

你是一位专业的Java后端开发工程师，擅长Spring Cloud微服务架构，
回答风格简洁专业，优先给出可运行的代码示例。

这会影响模型的语气、专业程度、回答侧重点，描述ai具体职责

2.2、指令说明

告诉模型”要做什么 / 不能做什么“，是最核心的部分

规则：
✅ 必须用中文回答
✅ 代码必须加注释
❌ 不要回答与编程无关的问题
❌ 不要捏造不存在的API

好的 Instructions 要包含：

• 正向规则（应该做）
• 负向规则（禁止做）
• 输出格式要求（JSON、Markdown等）

2.3、对话示例

用少样本学习（Few-shot） 的方式教模型想要的输出风格

示例输入：帮我解释什么是Redis缓存穿透
示例输出：
【定义】缓存穿透是指...
【危害】会导致...
【解决方案】1. Bloom Filter 2. 缓存空值

模型会从示例中学习格式和风格，比描述更直观有效

2.4、结构化输出

• 这个主要是为了去让模型去输出，传统的，适合模型去理解的内容

提示词版本

langchain官方版本

二、工具

Model 是大脑，Tools 是手脚，光有大脑只能”想”，有了手脚才能”做”

模型调用工具的流程

用户输入(input)
    ↓
  Model 分析三个问题：
    ① 需不需要调用工具？
    ② 调用哪个工具？
    ③ 工具结果够不够回答用户？
    ↓ (需要工具)          ↓ (不需要工具 / 结果够了)
  action(调用工具)        output(直接输出)
    ↓
  Tools 执行
    ↓
  observation(把结果返回给Model)
    ↓
  回到 Model 重新分析...

1.自定义工具

tool本质上就是一个函数，只不过是交给模型来决定”要不要调、怎么调”

1.1、基于@tool定义工具

# 定义工具
from langchain.tools import tool
# 通过装饰器定义工具作用描述
@tool("square_root", description="Calculate the square root of a number")
def tool1(x: float) -> float:
    return x ** 0.5

特点：

• 函数名 → 工具名
• docstring → 工具描述

1.2、函数名和文档注释描述工具

使用函数注释，来向模型表达工具作用，以及具体每个参数作用

1.3、Pydantic Model描述参数

from langchain_core.tools import tool
from pydantic import BaseModel, Field

# 1. 先定义参数结构
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="返回结果数量，默认5条")
    lang: str = Field(default="zh", description="语言，zh=中文，en=英文")

# 2. 绑定到工具上
@tool(args_schema=SearchInput)  # ← 指定参数模型
def search_web(query: str, max_results: int = 5, lang: str = "zh") -> str:
    """在网络上搜索信息"""
    return f"搜索'{query}'的{max_results}条{lang}结果..."

Pydantic Model 优势：

• Field(description=...) → 告诉模型每个参数是干什么的
• 自动做参数类型校验，传错类型会报错
• 参数复杂时结构清晰，便于维护

2.预定义工具

2.1、官网

https://docs.langchain.com/oss/python/langchain/tools#prebuilt-tools

https://docs.langchain.com/oss/python/integrations/tools

2.2、具体实现

https://app.tavily.com/home

使用tavily搜索工具

官方帮助文档

https://docs.langchain.com/oss/python/integrations/tools/tavily_search

普通用户每月可以有1000次调用机会

a.具体使用教程如下

• 配置环境变量

  

• 环境变量导入

  .env

  # tavily搜索工具
  TAVILY_API_KEY=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

• 安装依赖

  uv add langchain_tavily

• 初始化参数

  # 使用tavily作为web搜索工具
  from langchain_tavily import TavilySearch
  
  tool = TavilySearch(
      max_results=5,
      topic="general",
      # include_answer=False,
      # include_raw_content=False,
      # include_images=False,
      # include_image_descriptions=False,
      # search_depth="basic",
      # time_range="day",
      # include_domains=None,
      # exclude_domains=None
  )

• 作为简单工具调用搜索

  tool.invoke("蒸蚌是什么梗？")

  

b.配合智能体使用

• 官方集成

  # 创建智能体，使用预定义工具tavily
  agent = create_agent(
      model="deepseek-chat",
      tools=[tool],
      system_prompt="你是一个智能助手，你使用工具来解决用户问题。"
  )

  # 调用工具
  for chunk in agent.stream(
      {"messages": [HumanMessage(content="蒸蚌是什么梗？")]},
      stream_mode="updates"
  ):
      for step, data in chunk.items():
          print(f"step: {step}")
          print(f"content: {data['messages'][-1].content_blocks}")
          print()

  

c.优化token使用率

官方的 TavilySearchResults 工具有完整的参数列表，每次调用时 LLM 需要”读懂”这些参数，会额外消耗Token 和流量。对于简单业务，这些参数根本用不上，纯属浪费

# 第一步：用官方客户端做初始化（只配置必要参数）
tavily = TavilySearch(
    max_results=5,   # 最多返回5条搜索结果
    topic="general"  # 搜索主题为通用类型
)

这里只是创建底层客户端，还没有变成 LangChain 的 Tool

# 第二步：用 @tool 装饰器，把普通函数包装成 LangChain Tool
@tool
def web_search(query: str):
    """Search the web for information"""  # ← 这个 docstring 很关键！
    return tavily.invoke(query)

d.结构化输出

规定 Agent 回答时必须按照固定格式输出，而不是返回一段自由的文本字符串

核心依赖：Pydantic

from pydantic import BaseModel, Field

Pydantic 是 Python 的数据校验库，LangChain 大量使用它来做结构化约束。继承 BaseModel 的类，每个字都有类型检查，LLM 输出不符合格式会直接报错

class Reference(BaseModel):
    title: str = Field(description="The title of the web page cited in the answer")
    url: str   = Field(description="The url of the web page cited in the answer")
class AnswerInfo(BaseModel):
    answer: str              = Field(description="The final answer for user")
    reference: list[Reference] = Field(description="The web pages cited in the answer")

#创建智能体，使用预定义工具
tavilyagent = create_agent(
	model="deepseek-chat",
    tools=[web_search],
	system_prompt="你是一个智能助手，你使用工具来解决用户问题。",
    response_format=AnswerInfo
)

三、会话记忆

1.短期记忆实现

Agent短期记忆通过AgentStatus实现，Checkpoint用来记录每次的聊天节点，thread_id用来区分每次的聊天对话

一个 thread_id 对应一条时间线，时间线上每个节点执行后都有一个 Checkpoint，而 AgentStatus 反映的是最新 Checkpoint 的执行状态

1.1、thread_id

一次对话/任务的唯一标识符，类似于”会话 ID”

作用：

• LangGraph 用它来隔离不同用户/不同对话的状态
• 同一个 thread_id 的多次调用，会共享同一条状态历史线
• 不同 thread_id 之间完全独立，互不干扰

1.2、Checkpoint

LangGraph 在每个节点执行完之后，对当前完整状态的一次快照

包含内容：

• 当前的所有状态变量（messages、中间结果等）
• 执行到哪个节点了
• 下一步要执行哪个节点

作用：

• 断点续传： Agent 中断后可以从最后一个 Checkpoint 恢复，不用从头跑
• Human-in-the-loop： 在某个节点暂停，等人审批后再继续
• 时间旅行（Time Travel）： 可以回退到某个历史 Checkpoint 重新执行

1.3、AgentStatus

描述 Agent 当前处于哪个执行阶段的状态标识

常见状态值：

状态	含义
running	正在执行节点
interrupted	被 interrupt() 暂停，等待外部输入
error	执行过程中出错
end / finished	已到达终止节点，执行完毕

1.4、基于内存存储实现

导入依赖

InMemorySaver — 内存级别的 Checkpointer，把状态快照存在内存里（程序重启就丢失，适合开发测试用）

from langgraph.checkpoint.memory import InMemorySaver

创建 agent

agent = create_agent(
    "gpt-5",
    tools=[get_user_info],
    checkpointer=InMemorySaver(),
)

调用

from langchain.messages import HumanMessage

# 设定thread_id，作为会话标识
config = {"configurable": {"thread_id": "thread_1"}}

# 第一次调用，告知AI我的信息
response = agent.invoke(
    {"messages": [HumanMessage(content="你好，我叫虎哥，我最喜欢猫猫。")]},
    config # 调用时添加thread_id
)

print(response)

再次调用，即有记忆功能

# 第二次调用，询问我的信息，这次带上thread_id，唤起记忆
response = agent.invoke(
    {"messages": [HumanMessage(content="我最喜欢的动物是什么？")]},
    config # 调用时添加thread_id
)

print(response)

1.5、持久化存储数据库

具体实现

安装依赖

uv add langgraph-checkpoint-sqlite

导入依赖

from langgraph.checkpoint.sqlite import SqliteSaver

持久化实现

checkpointer = SqliteSaver(sqlite3.connect("resources/checkpoint.db", check_same_thread=False))
checkpointer.setup()

代码	说明
sqlite3.connect("resources/checkpoint.db", ...)	连接（或创建）一个本地 SQLite 数据库文件
check_same_thread=False	允许多线程访问同一个数据库连接，Agent 异步执行时必须加
SqliteSaver(...)	用这个数据库连接创建一个 SQLite 版的 Checkpointer
checkpointer.setup()	自动建表，在数据库里创建存储 Checkpoint 所需的表结构

对比上一节的 InMemorySaver：

• InMemorySaver → 存内存，重启丢失，适合开发调试
• SqliteSaver → 存磁盘，永久保留，适合生产使用

创建智能体

# 创建agent
agent = create_agent(
    "deepseek-chat",
    checkpointer=checkpointer,
)

直接调用

from langchain.messages import HumanMessage

# 设定thread_id，作为会话标识
config = {"configurable": {"thread_id": "thread_1"}}

# 第一次调用，告知AI我的信息
response = agent.invoke(
    {"messages": [HumanMessage(content="你好，我叫虎哥，我最喜欢猫猫。")]},
    config # 调用时添加thread_id
)

print(response)

config 单独定义好，后面两次调用复用同一个 config

用户输入：”我叫虎哥，我最喜欢猫猫”

Agent 处理完后，这次对话的完整状态被 存入 checkpoint.db，对应 thread_id = "thread_1"

2.记忆管理策略（summary）

多轮对话会不断积累历史消息，最终撑爆模型的上下文窗口（DeepSeek 上限 128K Token）,Token 越多，费用越高，速度越慢，甚至直接报错。所以需要对历史消息进行”瘦身”，一共有如下四种方式

官网链接

https://docs.langchain.com/oss/python/langchain/short-term-memory#common-patterns

2.1、总结策略具体实现

导入依赖

from langchain.agents.middleware import SummarizationMiddleware

初始化中间件

# 初始化总结中间件
middleware = SummarizationMiddleware(
    model="deepseek-chat",
    trigger=("messages", 3), #  触发时机，当消息数超过3时，进行总结
    keep=("messages", 1) #  保留的会话数，超过2条
)
# 初始化checkpointer
checkpointer = InMemorySaver()

创建agent

# 创建agent
agent = create_agent(
    model="deepseek-chat",
    middleware=[middleware],
    checkpointer=checkpointer,
)

总

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig

# 初始化checkpointer
checkpointer = InMemorySaver()
# 初始化中间件
middleware = SummarizationMiddleware(
    model="deepseek-chat",
    trigger=("messages", 3), #  触发时机，当消息数超过3时，进行总结
    keep=("messages", 1) #  保留的会话数，超过2条
)
# 创建agent
agent = create_agent(
    model="deepseek-chat",
    middleware=[middleware],
    checkpointer=checkpointer,
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
# 制造长会话历史
agent.invoke({"messages": [HumanMessage(content="你好，我是虎哥.")]}, config)
agent.invoke({"messages": [HumanMessage(content="我最喜欢的运动是乒乓")]}, config)
agent.invoke({"messages": [HumanMessage(content="我最喜欢的动物是猫猫")]}, config)
# 测试效果
final_response = agent.invoke({"messages": HumanMessage(content="你还记得我吗？")}, config)

2.2、中间件参数

参数一：model

用来生成摘要的模型，可以和主 Agent 的模型不同

# 技巧：用便宜的小模型生成摘要，节省成本
SummarizationMiddleware(
    model="openai:gpt-4o-mini"   # 摘要用小模型
)
# 而主 Agent 用 gpt-4o

参数二：trigger

支持三种触发方法

# 方式1：按消息条数触发（超过50条就摘要）
trigger=("messages", 50)

# 方式2：按 Token 数触发（超过3000 Token就摘要）
trigger=("tokens", 3000)

# 方式3：按比例触发（用到模型最大上下文的80%就摘要）
trigger=("fraction", 0.8)

# 方式4：多条件组合，任意一个满足就触发
trigger=[
    ("fraction", 0.8),   # 上下文用了80%
    ("messages", 100),   # 或消息超过100条
]

参数三：keep

摘要完成后，保留最近的 N 条消息不删除：

keep=("messages", 20)   # 保留最近20条（默认）
keep=("tokens", 3000)   # 保留最近3000 Token的消息
keep=("fraction", 0.3)  # 保留30%上下文容量的消息