Langchain 基础

# 大模型API

llm 与 chat_model

• llm 专门用于处理文本消息，仅接收文本内容，输出文本内容
• chat_model 对聊天做了优化，输出更为复杂。通常采用消息列表，其中包含角色信息

# 接口规范

基本都遵循 OpenAI 规范。

请求时路径 /chat/completions，使用 jwt 认证，请求体为 json，指明模型，消息数组（内部声明角色和内容），响应消息是否为流式。

可选请求参数 temperature: 采样温度，取之范围 $[0,2)$。温度越高，每个词的概率越均匀，文本多样性越高。温度越低，概率越集中。精确判断时调低，创意性任务时调高（1.2算是比较高）

• deepseek

  curl -X POST https://api.deepseek.com/chat/completions
    -H "Content-Type: application/json"
    -H "Authorization: Bearer <api key>"
    -d '{
      "model": "deepseek-chat",
      "messages": [{
          "role": "system",
          "content": "You are a helpful assistant."
      }, {
          "role": "user",
          "content": "Hello!"
      }],
      "stream": false
    }'
  
• qwen

  curl -X POST https://dashscope.aliyuncs.com/v1/chat/completions
    -H "Content-Type: application/json"
    -H "Authorization: Bearer <api key>"
    -d '{
      "model": "qwen-plus",
      "messages": [{
          "role": "system",
          "content": "You are a helpful assistant."
      }, {
          "role": "user",
          "content": "Hello!"
      }],
      "stream": false
    }'
  

因此可使用 OpenAI 的 SDK 进行请求。但注意，使用老版本的写法。OpenAI SDK 新版本的调用方式出现较大变动，但兼容老版本

# 新版本调用方式的写法，国内模型不可用
from openai import OpenAI
client = OpenAI()
response = client.responses.create(
    model='gpt-5.4',
    input='Hello!'
)
print(response.output_text)

# 老版本调用方式的写法
from openai import OpenAI
client = OpenAI(
    api_key='',
    base_url='https://api.deepseek.com'
)
response = client.chat.completions.create(  # 请求体 json 的字段作为参数传入
    model='deepseek-chat',
    message=[{
        "role": "system",
            "content": "You are a helpful assistant."
        }, {
            "role": "user",
            "content": "Hello!"
    }],
    stream=False
)
# 响应json
print(response.model_dump_json())
# 响应json路径以点的方式进行访问
print(response.choices[0].message.content)

# 会话记忆

大模型的服务是无状态的。每次请求是一次会话，不同会话相互独立，不具备记忆能力

因此，需要在请求体的 messages 数组中拼接所有历史消息

# LangChain

agent 开发框架

常用组件

• LangSmith: 监控运行细节、基于测试集测试评估智能体、优化智能体提示词、一键部署agent
• LangSmith Fleet: 0 代码开发 agent

什么是 agent？

智能体或代理。能够感知环境，进行推理、自主决策并采取行动以实现特定目标的智能系统

执行能力更强，可以操作软件

能思考任务步骤，选择合适工具，执行工具调用，根据结果调整策略，直到完成任务

智能体的部署：

• 使用 LangSmith（需要付费）
• 将 agent 接入 web 服务进行部署

# 智能体入门

在没有 Agent 时，大模型只能基于训练数据回答问题。例如在如果不打开搜索功能，询问大模型明天的天气，则只能从训练集中获取过去训练过的天气信息

测试编写一个查询天气的智能体

• 安装依赖: uv add langchain 再安装对应大模型的依赖，例如

  • langchain-deepseek
  • langchain-openai
  • langchain-anthropic
  • langchain-google-genai
  • langchain-ollama
  • langchain-huggingface
  • langchain-aws

• 定义工具

  from langchain.tools import tool
  # 装饰器 - 声明该函数是一个工具，后续请求大模型时，会自动生成工具的请求 json 结构携带到请求体的 tools 字段中
  @tool
  def getWeather(location: str) -> str:
      # 测试时先返回固定天气
      return f'Current weather in {location} is sunny'
  

• 创建 Agent。通过环境变量声明 api key 等内容，langchain 会自动加载。如 deepseek 需要声明环境变量 DEEPSEEK_API_KEY

  from langchain.agents import create_agent
  # 声明使用的模型 和 可调用的函数
  agent = create_agent('deepseek-chat',  tools=[getWeather])
  # 向大模型发送请求时的可选参数 tools, 用于告知模型有哪些工具可以调用
  

• 调用 agent

  response = agent.invoke({
      'message': [{
          'role': 'user',
          'content': '杭州今天天气怎么样？'
      }]
  })
  

智能体工作流程

• 用户提问

• 模型分析(思考 Reasoning): 是否需要调用工具，调用什么工具

  • 请求体 json 中的可选参数 tools 用于描述当前有哪些工具可以调用

• 调用工具(行动 Action): 执行 思考 Reasoning 阶段制定的策略

  • 大模型响应的 choices 中除了拥有 message 文本信息外，还拥有 tool_calls 用于告知客户端 ai 想调用哪个工具。
  • langchain 作为客户端解析该结果并调用对应的工具，将调用结果返回给大模型

    for message in reponse['message']:  # 循环遍历消息列表，即对话历史记录
        print(message.model_dump_json(indent=2))
    

• 感知分析(观察 Observation): 分析工具执行结果，是否足以完成目标，还是要继续调用工具

• 生成结果

以上流程即为 ReAct

全称 Reasoning + Acting（推理 + 行动），智能体的核心思考和行动框架，是 Agent 像人类一样 思考问题、制定策略、执行行动、验证结果 的关键逻辑

不直接回答问题，而是按自然语言思考过程，选择工具并调用，一步步结合复杂问题。以上流程描述体现了 ReAct 的核心

# 模型初始化

• 了解 langchain 对应不同大模型的依赖和模型的 api
  • langchain-deepseek ChatDeepSeek
  • langchain-openai ChatOpenAI
  • langchain-anthropic ChatAnthropic
  • langchain-google-genai ChatGoogleGenerativeAI
  • langchain-ollama ChatOllama
  • langchain-huggingface HuggingFaceEndpoint
  • langchain-aws ChatBedrock
• 实例化对应的模型。
  • 以 OpenAI 为例

    from langchain_openai import ChatOpenAI
    model = ChatOpenAI(model='gpt-5.2')
    
  • 1.1 版本后的新方式。简化多模型协作时导入过多

    from langchain.chat_models import init_chat_model
    model = init_chat_model(model='gpt-5.2')
    
• 开发步骤
  • 安装依赖: langchain 与 对应模型
  • 配置环境变量，不同模型的环境变量要求不同，需要查阅 langchain 官方文档，例如 deepseek 需要 DEEPSEEK_API_KEY
  • 初始化模型
    • 使用 init_chat_model。注意是否支持，例如 qwen 不被支持，由于兼容 openai，可通过伪装成 openai 的方式

      # 自定义环境变量 QWEN_BASE_URL 和 QWEN_API_KEY
      model = init_chat_model(
          model='qwen-max',  # 声明模型名
          model_prvoider='openai',  # 声明供应商
          base_url=os.getenv('QWEN_BASE_URL'),
          api_key=os.getenv('QWEN_API_KEY')
      )
      print(type(model))  # 由于类型是根据 prvoider 确定，此处 type 就是 ChatOpenAI
      
    • 手动定义。对于官方不兼容的模型，查看 langchain 社区是否支持该模型
      • 先安装 uv add langchain-community
      • 再寻找模型官方提供的 SDK 进行安装，比如 qwen 的 dashscope
      • 就可以通过 langchain_community 导入并使用该模型

        from langchain_community.chat_models.tongyi import ChatTongyi
        
        model = ChatTongyi(model='qwen-max')
        

# 模型调用

使用 model.invoke()（阻塞式） / model.stream()（流式），省略模型名、流式传输、温度等每次固定的参数，只需传输消息列表即可

response = model.invoke([{
    "role": "system",
    "content": "You are a helpful assistant."
}, {
    "role": "user",
    "content": "Hello!"
}])

如果只考虑单次提问，且无系统提示词，可直接使用 model.invoke('Hello!')

由于流式结果是生成器，访问流式结果需要 for in 循环遍历

• stream 函数的参数 stream_mode='message: 模型返回的消息内容采用流式方式

  for chunk in stream:
      # 只打印响应体中的 content 字段内容，且不换行
      print(chunk.content, end='', flush=True)
  
• stream 函数的参数 stream_mode='values: 对话消息历史采用流式方式。即每次得到的是当前轮次的的消息列表数组，数组在每轮中会添加最新消息

  for chunk in stream:
      # 获取当前轮次的消息数组
      current_msg_list = chunk['messages']
      # 从当前轮次消息数组中获取当前最新添加的消息
      latest_msg = current_msg_list[-1]
      if latest_msg.content:
          # 打印出消息种类(以此判断发送消息的角色) 和 消息内容
          print(type(latest_msg).__name__, latest_msg.content)
      try:
          # 仅 AIMessage 会携带需要调用工具的内容
          if latest_msg.tool_calls:
              # 打印出工具调用信息，可能会一次要求调用多个工具
              print(f"tools: {[tc['name'] for tc in latest_msg.tool_calls]}")
      except AttributeError:  # 仅 AIMessage 会携带需要调用工具的内容，其他消息无该属性
          pass
  

直接调用模型就是使用 invoke 或 stream 方法。而大多数情况下，不是直接调用大模型，而是在智能体中使用大模型

• 创建智能体

  from langchain.agents import create_agent
  # 方法 1: 传入创建好的 model
  agent = create_agent(model=model)
  # 方法 2: 指定模型名称，内部自己执行 init_chat_model 创建模型。因此也需要 langchain 官方支持模型的名称
  agent = create_agent(model='deepseek-chat')
  

• 调用智能体。同调用大模型一样，使用 invoke 或 stream 方法

  response = agent.invoke({
      # 智能体的请求体中可用的参数众多，传入消息时需要 `message` 字段明确说明传入的是消息
      'message': [{
          "role": "system",
          "content": "You are a helpful assistant."
      }, {
          "role": "user",
          "content": "Hello!"
      }]
  })
  
  messages = agent.stream({
      # 智能体的请求体中可用的参数众多，传入消息时需要 `message` 字段明确说明传入的是消息
      'message': [{
          "role": "system",
          "content": "You are a helpful assistant."
      }, {
          "role": "user",
          "content": "Hello!"
      }],
      # 指明流的模式。message 模式是常规的连续吐字模式
  }, stream_mode='message')
  for token, metadata in messages:
      # 检查是否有响应内容
      if token.content:
          print(token.content, end='', flush=True)
  # stream_mode='values'时，遍历得到的是历史消息数组。
  # 可观察到每次消息数组的变化，每次新增的内容。chunk['messages'][-1] 即是每次新增的消息
  

# 消息

有的消息使用 str，有的消息使用 list[dict]，此处介绍 langchain 常用的消息格式

消息存在多种类型类型

在 langchain 中，调用端与大模型间来回传输的消息，都被封装为 BaseMessage，且分为多个子类，每个子类对应不同角色类型的消息

• SystemMessage Role: system
• HumanMessage Role: user
• AiMessage Role: assistant 即模型返回的消息
• ToolMessage Role: tool 即工具调用后响应给大模型的消息

使用下列示例代码，观察传输的消息

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def get_weather(location: str) -> str:
    return f'Current weather in {location} is sunny'

# 使用 langchain 官方支持的模型名称直接创建智能体，而不是传入模型实例的方式
agent = create_agent(model='deepseek-chat', tools=[get_weather])

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

response.agent.invoke({
    'message': [
        # 等效为 {'role': 'system', 'content': '你是一个 AI 助手'}
        SystemMessage(content='你是一个 AI 助手'),  # 关键字 content 可以省略
        # 等效为 {'role': 'user', 'content': '你好，我是用户'},
        HumanMessage('你好，我是用户')
        # 等效为 {'role': 'assistant', 'content': '你好，用户！请问有什么需要帮忙的？'},
        AIMessage('你好，用户！请问有什么需要帮忙的？')
        # 等效为 {'role': 'user', 'content': '上海今天天气怎么样？'}
        HumanMessage('上海今天天气怎么样？')
    ]
})
print(response)
# 遍历消息
for message in response['messages']:
    message.pretty_print()

# 得到以下输出：

# ============== System Message ==============
# 你是一个 AI 助手
# ============== Human Message ==============
# 你好，我是用户
# ============== AI Message ==============
# 你好，用户！请问有什么需要帮忙的？
# ============== Human Message ==============
# 上海今天天气怎么样？
# ============== AI Message ==============
# 我来帮你查询一下上海今天的天气情况
# Tool Calls
#   get_weather (call_00_xxxx)
#  Call ID: call_00_xxxx
#   Args:
#     location: 上海
# ============== Tool Message ==============
# Name: get_weather
#
# Current weather in 上海 is sunny
# ============== Ai Message ==============
# 根据查询结果，上海进行的天气是晴朗的

还可以进一步简化。此时不同点是，Message类会在实例化时直接得到实例，元组则是 langchain 读取到该元组后动态转换成 Message 示例

# {'role': 'system', 'content': '你是一个 AI 助手'}
# SystemMessage(content='你是一个 AI 助手')
('system', '你是一个 AI 助手')

# {'role': 'user', 'content': '你好，我是用户'}
# HumanMessage('你好，我是用户')
('human', '你好，我是用户')

# {'role': 'assistant', 'content': '你好，用户！请问有什么需要帮忙的？'}
# AIMessage('你好，用户！请问有什么需要帮忙的？')
('ai', '你好，用户！请问有什么需要帮忙的？')

# 该方式便于链式调用时注入变量，因为是动态转换的，链式调用模型时才会进行替换并转换
('system', '今天天气是{weather}')

除了发送方不一致导致的 Base Message 子类不同。Langchain 还支持向模型发送多模态消息（Multimodal），比如图片、音频、视频、文本等。

但需要提前确认所使用的模型是否支持多模态

查看 langchain 官网的 Integrations 接入页面的 Chat models 查看模型是否支持多模态（模型的 Multimodal 一栏是否打勾）

比如 deepseek、ollama 等不支持多模态

下面以图片为例，说明如何发送图片消息

from langchain.chat_models import init_chat_model

model = init_chat_model(
    model='qwen3.5-plus',  # 根据 qwen 官网，该模型支持图片、视频的输入
    model_provider='openai',  # 由于 langchain 不支持 qwen，但 qwen 兼容 openai。因此伪装成 openai
    base_url=os.getenv('DASHSCOPE_BASE_URL'),
    api_key=os.getenv('DASHSCOPE_API_KEY')
)
# 由于是非 langchain 官方支持的模型，使用模型实例构建智能体
agent = create_agent(model=model)

# 方式一：以图片链接的形式发送图片，内容需要是数组
# {'role': 'user', 'content': [
#     {'type': 'text', 'text': '图片描述'},
#     {'type': 'image', 'url': 'https://xxxx'}
# ]}
message = HumanMessage([  # content 直接传入内容数组
    {'type': 'text', 'text': '图片描述'},
    {'type': 'image', 'url': 'https://xxxx'}
])
# 方式二：将图片转换成 base64 编码 - 此方式请求体过大，会浪费 token
# {'role': 'user', 'content': [
#     {'type': 'text', 'text': '图片描述'},
#     {'type': 'image', 'base64': 'xxxx', 'mime_type': 'image/jpeg'}
# ]}
img_b64 = base64.b64encode(image_bytes).decode('utf8')
message = HumanMessage([  # content 直接传入内容数组
    {'type': 'image', 'base64': img_b64, 'mime_type': 'image/jpeg'},
    {'type': 'text', 'text': '描述一下图片内容'}
])

# 传入消息数组，注意流式调用需再传入流式模式
stream = agent,stream({'message': [message]}, stream_mode='message')
for chunk, metadata in stream:
    if chunk.content:
        print(chunk.content, end='', flush=True)

# 提示词工程

所有发送给大语言模型的消息都是提示词

消息数组中发送的 SystemMessage 就是系统提示词。

最简单的发送方法时在消息数组内添加 SystemMessage，后续聊天内容追加在数组内

由于系统提示词作为系统指令，每次都出现在消息数组中，有些麻烦。langchain 推荐在创建智能体时使用 system_prompt 参数而不是消息数组内的 SystemMessage 来传递系统提示词

from langchain.agents import create_agetn
agent = create_agent(model='deepseek-chat', system_prompt='你是一个智能助手，以简短的中文回答用户问题')

# 后续向智能体传递消息时，消息数组内无需 SystemMessage

提示词工程，指通过优化提示词使模型输出的结果更符合业务需要的过程。所以提示词需要反复调试，不断迭代。

系统提示词需要包含的部分，按以下顺序排列

• 身份角色：描述AI指责、沟通风格、总体目标
• 指令说明：（身份角色的补充）指导模型如何生成所需响应，应遵循的规则，应该做什么，不能做什么。指定任务的步骤，详细说明
• 对话示例：提供输入示例和期望输出
• 背景信息（可选）：额外补充信息。使用参考文本，让模型用参考文本作答。让模型从指定文本中筛选出答案，而不是自己造答案。如 RAG 中的知识库数据

推荐编写格式是 md 或 xml，使用分隔符标明输入的不同部分

• 角色和指令：只设置身份角色可能不符合需求。比如明确说明指令部分，简单告知只需要做什么，不需要做什么，可以节省 token

  # 角色
  - 你是一个编程助手，给用户提供 python 代码
  # 指令
  - 不要返回说明信息，仅返回代码内容
  - 仅保留关键代码的注释
  
• 对话示例：Few-Shot examples 是一种为模型提供示例的方法（Few-Shot Learn, 少样本学习）。提供的示例中，答案风格也会影响 AI 输出。简洁的答案示例可以避免输出内容过多，实现节省 token

  # 角色
  - 你是一个地理老师，告知用户地理相关问题的答案
  # 示例
  user: 中国的首都是哪座城市
  assistant: 北京
  
  user: 山东的省会是哪里
  assistant: 济南
  
• 结构化输出：很多时候需要 ai 返回结构化内容，如 json / xml 等。便于代码操作。可通过指令与示例详细说明

  • 常规方式

    # 角色
    - 你是一个地理老师，告知用户地理相关问题的答案
    # 指令
    - 务必保证以 json 格式输出，携带省份与城市信息
    # 示例
    user: 少林寺在哪个城市
    assistant: 
    {
        "province": "河南",
        "city": "郑州"
    }
    
    user: 泰山在哪里
    assistant:
    {
        "province": "山东",
        "city": "泰安"
    }
    

  • langchain 中对结构化输出进行了优化。由于对接了 pydantic，可使用 pydantic 编写大模型输出的结构格式

    from pydantic import BaseModel
    class City(BaseModel):
        province: str
        city: str
    
    from langchain.agents import create_agent
    agent = create_agent(
        model='deepseek-chat',
        system_prompt='你是一个地理老师，告知用户地理相关问题的答案',
        response_format=City
    )
    response = agent.invoke({'messages': [HumanMessage('黄山在哪里？')]})
    # 响应中存在 structured_response 字段，即为结构化类的实例
    print(response['structured_response'])  # 此处响应的是一个 City 类的实例
    

# 提示词模版

langchain 提供了提示词模型相关类，用于协助优化提示词。

构建模版拥有 from_template（单条消息）与 from_messages（多条消息，ChatPromptTemplate 专属）

模版实例拥有 format(生成字符串) 与 invoke(生成对象，用于构建链) 方法，注意区别

• PromptTemplate 通用提示词模版，一般用于 llm。以下是一个 llm 的示例

  from langchain_core.prompts import PromptTemplate
  prompt_template = PromptTemplate.from_template(
      '我的邻居姓{lastname}，刚生了{gender}，帮忙起名字，请简略回答'
  )
  # format 方法，得到提示词字符串。需传入占位符变量 = 值
  prompt_text = prompt_template.format(lastname='张', gender='女儿')
  print(llm.invoke(input=prompt_text))
  
  • 相比于 f-string，PromptTemplate 可使用 langchain 的链式调用，将提示词通过 chain 链交给模型或 agent

    prompt_template = PromptTemplate.from_template(
        '我的邻居姓{lastname}，刚生了{gender}，帮忙起名字，请简略回答'
    )
    # 生成 chain，先将提示词对象放入模型
    chain = prompt_template | model
    # 动态注入的方式进行调用
    chain.invoke(input={'lastname': '张', 'gender': '女儿'})
    

• FewShotPromptTemplate 示例模版

  from langchain_core.prompts import FewShotPromptTemplate
  FewShotPromptTemplate(
      examples=None,  # list[dict] 示例数据
      example_prompt=None,  # 每个示例的提示词格式
      prefix=None,  # 组装提示词，示例前的提示词内容
      suffix=None,  # 组装提示词，示例后的提示词内容
      input_variables=None  # list[str]，注入提示词的变量名列表
  )
  
  # 每个示例的提示词格式模版
  example_template = PromptTemplate.from_template('单词：{word}，反义词：{antonym}')
  # 示例数据，会带入示例提示词，组成多个示例
  example_data = [
      {'word': '大', 'antonym': '小'},
      {'word': '上', 'antonym': '下'},
  ]
  few_shot_prompt = FewShotPromptTemplate(
      examples=example_data,
      example_prompt=example_template,
      prefix='给出定于的反义词，下面是一些示例',
      suffix='基于以上示例，回答我：{input_word}的反义词是？',
      input_variables=['input_word']  # 变量名列表
  )
  
  # invoke 方法，得到 PromptValue 对象（保留占位符，用于构建链），再调用 to_string 得到字符串。需传入占位符变量的字典
  prompt_text = few_shot_prompt.invoke(input={'input_word': '左'}).to_string()
  # 查看最终组成的提示词
  print(prompt_text)
  # 传统方式调用
  llm.invoke(input=prompt_text)
  

• ChatPromptTemplate 历史对话信息

  from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
  chat_template = ChatPromptTemplate.from_messages([
      ('system', '...'),
      ('ai', '...'),
      MessagesPlaceholder('history'),  # 历史消息占位符
      ('human', '....')
  ])
  history_data=[('human', '...'), ('ai', '...')]
  # 将占位符替换成真实值
  chat_template.invoke({'history': history_data})
  

# 链

组件串联，将前一项组件的输出作为后一项组件的输入。

Runnable 的子类拥有 invoke 与 stream 方法，比如消息模版，模型都是 Runnable 子类

Runnable 子类对象可以入链，因此可以编写 chain = prompt_template | model

prompt_template = PromptTemplate.from_template(
    '我的邻居姓{lastname}，刚生了{gender}，帮忙起名字，请简略回答'
)
chain = prompt_template | model
# 在此例中，dict 作为链的原始输入，给到 prompt_template，输出 PromptValue 实例。该实例又作为 model 的输入，给到模型对象
chain.invoke(input={'lastname': '张', 'gender': '女儿'})

本质上，即 Runnable 重写了 __or__ 方法，返回 RunnableSquence，每一项都会追加到 RunnableSquence 中

# 输出解析器

将 chat_model 回复的结果再交给 chat_model 进行提问，由于 chat_model 的输出是 AIMessage， chat_model 第二次收到消息是 AIMessage，此时会产生异常

即 chain = prompt | model | model 这个链在调用 invoke / stream 时会报错，仅接收 提示词实例 / str / 消息列表

可使用输出解析器处理模型的输出，使其可以作为模型的输入。例如 StrOutputParser 可将模型输出处理成简单字符串

• StrOutputParser: 将传入的 Message 实例转换成 str

  from langchain_core.output_parsers import StrOutputParser
  parser = StrOutputParser()
  chain = prompt | model | parser | model
  

链中存在不止一次模型，即为多模型执行链。在实际情况下，模型的输出内容不会直接作为模型的输入，而是先进行数据处理，放入提示词模版，再提交给模型

由提示词模版通常需要 dict 类型，因此第一个模型的输出需要转换为 dict 类型，可使用 JsonOutputParser

• JsonOutputParser

  from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
  str_parser = StrOutputParser()
  json_parser = JsonOutputParser()
  from langchain_community.chat_models.tongyi import ChatTongyi
  model = ChatTongyi(model='qwen3-max')
  from langchain_core.prompts import PromptTemplate
  first_prompt = PromptTemplate.from_template(
      '我邻居姓{lastname}，刚生了{gender}。请起一个名字。要求返回json格式，key是name，value是结果。严格遵循格式要求'
  )
  second_prompt = PromptTemplate.from_template('姓名{name}，帮我解析含义')
  chain = first_prompt | model | json_parser | second_prompt | model | str_parser
  res = chain.stream({'lastname': '张', 'gender': '女孩'})
  for chunk in res:
      print(chunk, end='', flush=True)  # 由于最终经过了 StrOutputParser，输出就是字符串内容，无需 .content 等处理
  

有时需要结构化输出，除了使用 response_format 参数传入 pydantic 类得到类实例外，也使用 JsonOutputParser 实现输出 json 内容。此方案需要在提示词中告知模型输出的格式

# 自定义函数

如果想在链中加入自定义函数（比如打印链中某个步骤的输出，方便调试等），需要调用 RunnableLambda

例如 JsonOutputParser 的例子，也可使用自定义函数将模型的输出构造成 dict

from langchain_core.output_parsers import StrOutputParser
str_parser = StrOutputParser()
from langchain_community.chat_models.tongyi import ChatTongyi
model = ChatTongyi(model='qwen3-max')
from langchain_core.runnables import RunnableLambda
# 自定义函数，使用 RunnableLambda 包裹
my_func = RunnableLambda(lambda ai_msg: {'name': ai_msg.content})
from langchain_core.prompts import PromptTemplate
first_prompt = PromptTemplate.from_template(
    '我邻居姓{lastname}，刚生了{gender}。请起一个名字。要求只返回结果，不需要额外信息'
)
second_prompt = PromptTemplate.from_template('姓名{name}，帮我解析含义')
chain = first_prompt | model | my_func | second_prompt | model | str_parser
# 等效为以下写法，底层会自动转换成 RunnableLambda。直接将函数传入
# first_prompt | model | (lambda ai_msg: {'name': ai_msg.content}) | second_prompt | model | str_parser
res = chain.stream({'lastname': '张', 'gender': '女孩'})
for chunk in res:
    print(chunk, end='', flush=True)

# 由于可以直接将函数传入，普通函数在保证输入与输出时，可直接将函数名放入链中。如调试时使用的提示词模版的函数
def print_prompt(prompt):
    print(prompt.to_string())
    print('='*20)
    return prompt  # 保证该函数有输入交给链中的后续环节

# 工具

一个完整的 Agent 至少要包含两个关键的部分。 模型是智能体的大脑，负责分析。而工具是智能体的手脚，负责执行任务，与外界交互

• 自定义工具：本身就是一个可调用的函数，但是该函数是交给 ai 调用的函数。需要ai知道的信息有，工具名，功能描述，入参信息
  • 基于 tool 描述工具。使用 langchain 中的 @tool 装饰器

    # 普通函数
    def sqrt(x: float) -> float:
        return x ** 0.5
    
    # 将普通函数改造为工具
    from langchain_core.tools import tool
    #      工具名         函数的描述信息，由于参数就一个，且描述中有说明计算一个数的平方根，此处参数信息可省略
    @tool('square_root', description='Calculate the square root of a number')
    def sqrt(x: float) -> float:
        return x ** 0.5
    
  • 采用函数名和文档注释描述工具: 函数名作为工具名（因此函数名需要见文知意），描述信息和入参信息可放入函数的"""注释中

    @tool
    def square(x: float) -> float:
        """Calculate the square root of a number"""
        return x ** 0.5
    
  • 对于复杂参数，使用 Pydantic Model 描述参数

    from pydantic import BaseModel, Field
    from typing import Literal
    # 所有参数的描述信息
    class WeatherInput(BaseModel):
        #        参数类型       参数描述
        location: str = Field(description='City name or coordinates')
        #        枚举类型的说明方式                         参数默认值           参数描述
        units: Literal['celsius', 'fahrenheit'] = Field(default='celsius', description='Temperature unit preference')
        include_forecast: bool = Field(default=False, description='Include 5-day forecast')
    
    #         声明入参说明的 model
    @tool(args_schema=WeatherInput)
    # 正常编写原始的函数名和参数，当作入参 model 不存在（即不要省略默认值和类型等信息）
    def get_weather(location: str, units: str='celsius', include_forecast: bool = False) -> str:
        # 函数描述还是正常编写
        """Get current weather and optional forecast"""
        temp = 22 if units == 'celsius' else 72
        result = f'Current weather in {location}: {temp} degrees {units[0].upper()}'
        if include_forecast:
            result += '\nNext 5 days: Sunny'
        return result
    
• 预定义工具：根据文档open in new window，有搜索，代码执行，生产力，浏览器操作，数据库等工具
  • 示例：Tavily 使用教程。根据文档open in new window，有自己一套 SDK，但 langchain 也集成了该工具

    # 这里使用 langchain 文档中 langchain 集成的 SDK，而不是 tavily 自己的 SDK
    from langchain_tavily import TavilySearch
    search_tool = TavilySearch(
        max_results=5,
        topic='general',
        # 参数信息查看 tavily 文档
    )
    # 使用时将 search_tool 放入 tools 数组内
    
  • Tavily 官方工具改为自定义工具。由于 Tavily 参数众多，ai 调用时很多 token 浪费在了没有使用的参数描述上，会增加 token 消耗

    tavily = TavilySearch(max_results=5, topic='general')
    
    # 使用 langchain 封装的 SDK 或 tavily 自己的 SDK 封装一个少参数的函数，做成工具
    @tool
    def web_search(query: str):
        """Search the web for information"""
        return tavily.invoke(query)
    
  • 定义结构化输出实现来源输出

    from pydantic import BaseModel, Field
    # Angent 回答内容引用的网页信息
    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')
    # Agent 回答内容
    class AnswerInfo(BaseModel):
        answer: str = Field(description='The final answer for user')
        reference: list[Reference] = Field(description='The web pages cited in the answer')
    
    agent = create_agent(
        model='deepseek-chat',
        tools=[web_search],
        system_prompt='你是一个智能助手，请使用工具来解决用户问题'
        response_format=AnswerInfo
    )
    resp = agent.invoke({'message': [HumanMessage(content='衡山在哪里')]})
    print(response['structured_response'])
    

# 记忆

llm、chat_model、agent 默认都是没有记忆的，即每次调用时，都是一次全新的对话，不记得之前发送了哪些消息。

最简单的方法是维护一个消息数组，每次将新的消息追加到其中。维护消息数组的流程比较固定化，官方也提供了内置工具用于处理记忆问题

# model 记忆

在与实例化的 llm 或 chat_model 进行对话时，模型被视作一个最简状态机（只有 1 个模型节点，没有 Agent 的工具调用循环）

• 旧写法，现在已不在维护。使用 langchain_core.runnables.history 中的工具

  • RunableWithMessageHistory 在原有链的基础上，创建带有历史记录的新链（提供记录历史的功能）

    from langchain_core.runnables.history import RunnableWithMessageHistory
    
    conversion_chain = RunnableWithMessageHistory(
        runnable_obj,  # 被附加历史消息的 Runnable(chat_model/llm) / chain
        get_history_obj_func,          # 获取指定会话 ID 的历史会话函数 - 用于区分不同会话的函数，传入 ID 得到对应的会话历史实例对象。
        input_message_key='input',  # 用户输入消息在模版中的占位符变量名
        history_message_key='chat_history'  # 声明历史消息在模版中的占位符变量名
    )
    
    # 关于获取指定会话 session ID 的历史会话函数：
    #   用于区分不同会话的函数，传入 session ID: str，返回对应的会话历史实例对象
    #   session_id 即是调用模型时手动指明的配置项 session_id
    #   RunnableWithMessageHistory 需要一个函数用于获取指定 ID 的会话
    
  • InMemoryChatMessageHistory 历史记录都存储到内存中

    from langchain_core.messages import HumanMessage
    from langchain_core.chat_history import InMemoryChatMessageHistory
    from langchain_core.runnables.history import RunnableWithMessageHistory
    from langchain_openai import ChatOpenAI
    
    model = ChatOpenAI(model="gpt-4o")
    store = {}  # 存放不同 session 的历史消息
    
    # 区分会话函数，传入 ID，返回会话历史实例对象
    def get_session_history(session_id: str):
        if session_id not in store:  # 该会话还不存在
            store[session_id] = InMemoryChatMessageHistory()
        return store[session_id]
    
    # 1. 不使用提示词模版普通的调用，无需 message_key 参数
    with_history = RunnableWithMessageHistory(  # 实例化 history，此处包装的 model，还需传入区分会话的函数
        model,
        get_session_history
    )
    config = {"configurable": {"session_id": "user_1"}}
    print(with_history.invoke([HumanMessage(content="你好，我叫张三")], config=config))
    
    # 2. 使用通用提示词模版
    prompt = PromptTemplate.from_template('你需要更具对话历史回应用户问题。对话历史：{chat_history}。用户问题：{input}，请给出回应')
    base_chain = prompt | model | StrOutputParser()
    conversion_chain = RunnableWithMessageHistory(  # 基于无记忆链，构造一个可以记录历史消息的链
        base_chain,
        get_session_history，
        input_message_key='input',  # 用户输入消息在模版中的占位符变量名
        history_message_key='chat_history'  # 声明历史消息在模版中的占位符变量名
    )
    config = {"configurable": {"session_id": "xiaoming"}}
    # 注意，此时用户提示词必须使用 dict。输入内容是 {'<用户提示词占位符>': '<用户内容>'}
    print(conversion_chain.invoke({'input': '小明爱吃苹果'}, config))  # 
    print(conversion_chain.invoke({'input': '小明爱吃什么水果'}, config))
    # 通过打印提示词可发现实际流程
    # 第一次调用时，RunnableWithMessageHistory 实例会调用获取会话历史实例对象函数
    #   由于第一次消息，函数返回一个空的 InMemoryChatMessageHistory 对象，此时 chat_history 会被替换成空 list。本次的用户输入正常替换到对应占位符
    # 第二次调用时，调用函数得到有内容的 InMemoryChatMessageHistory 对象，此时会将消息数组替换到 chat_history 处。本次的用户输入正常替换到对应占位符
    
    # 3. 聊天提示词模版 - 与上一步的通用提示词模版内容等效
    ChatPromptTemplate.from_messages([
        ('system', '你需要更具对话历史回应用户问题'),
        MessagePlaceHolder('chat_history'),
        ('human', '用户问题: {input}')
    ])
    
  • 基于 BaseChatMessageHistory 实现一个消息历史落盘的历史记录存储。
    • 需要实现的三个方法
      • add_messages: 同步模式，指定会话中添加消息
      • messages: 同步模式，获取指定会话的消息
      • clear: 同步模式，清除指定会话的消息历史
    • 基于不同 session 存储不同 json 文件的历史消息存储器的实现

      import os, json
      from langchain_core.chat_history import BaseChatMessageHistory
      # list[BaseMessage] 与 list[dict] 互相转换的方法
      from langchain_core.messages import messages_from_dict, message_to_dict
      
      class FileChatMessageHistory(BaseChatMessageHistory):
          def __init__(self, session_id: str, storage_path: str):
              self.session_id = session_id   # 调用时指定的 session_id
              self.storage_path = storage_path  # 消息历史 json 文件的存储目录
              self.file_path = os.path.join(self.storage_path, self.session_id)  # 消息历史 json 文件路径
              os.makedirs(os.path.join(self.storage_path), exist_ok=True)  # 检测文件目录
          
          @property
          def messages(self) -> list[BaseMessage]:
              """获取指定会话的消息列表"""
              try:
                  with open(self.file_path, 'r', encoding='utf-8') as f:
                      messages_data = json.load(f)
                  return messages_from_dict(messages_data)  # list[BaseMessage] -> list[dict]
              except FileNotFoundError:
                  return []
          
          def add_message(self, messages: Sequence[BaseMessage]) -> None:
              """指定会话中追加最新消息"""
              all_messages = list(self.messages)
              all_messages.extend(messages)  # 通常需要追加的是一问一答
      
              serialized = [message_to_dict(message) for message in all_messages]  # 转换成 list[dict]，后续可 json.dump
              with open(self.file_path, 'w', encoding='utf-8') as f:
                  json.dump(serialized, f)
          
          def clear(self) -> None:
              """清空指定会话内的消息记录"""
              with open(file_path, 'w', encoding='utf-8') as f:
                  json.dump([], f)
      
      # 区分会话函数，传入 ID，返回会话历史实例对象
      def get_session_history(session_id: str):
          # 由于 FileChatMessageHistory 内部已经处理了会话不存在的情况，直接返回
          return FileChatMessageHistory(session_id, './chat_history/')
      

• 更现代的写法

  from langchain_openai import ChatOpenAI
  from langgraph.graph import StateGraph, START, MessagesState
  from langgraph.checkpoint.memory import MemorySaver
  
  # 1. 初始化模型
  model = ChatOpenAI(model="gpt-4o-mini")
  
  # 2. 定义唯一的一个纯模型调用逻辑（没有 Agent，不调用 Tool）
  def call_model(state: MessagesState):
      # state["messages"] 自动包含由 checkpointer 管理的历史消息
      response = model.invoke(state["messages"])
      # 返回的新消息会自动通过 reducer 追加到 messages 列表中
      return {"messages": [response]}
  
  # 3. 构造极简图：START -> call_model
  workflow = StateGraph(state_schema=MessagesState)
  workflow.add_node("model", call_model)
  workflow.add_edge(START, "model")
  
  # 4. 挂载持久化检查点并编译
  app = workflow.compile(checkpointer=MemorySaver())
  

# agent 记忆的分类

langchain 自带记忆管理功能（本质上是 langgraph 自带的，langchain 基于 langgraph）。agent 的记忆分两类

• 短期记忆：当前任务或会话的上下文。例如：对话历史，查询结果，任务状态
• 长期记忆：跨任务或会话的经验与知识。例如：知识库，用户偏好，失败经验

# checkpointer

其中，短期记忆是通过 AgentState 实现的，会话历史是 AgentState 的一部分。会话历史是 AgentState 已经实现的，其他短期记忆内容需要自行实现 AgentState。

Langchain 提供了 Checkpointer 对象来保存 AgentState，每次用户与 AI 交互都会生成一个快照，记录为 checkpointer

同一个会话的多个 checkpointer 形成一个组，用同一个 thread_id 标记（实际含义同会话 id）

如果想给智能体添加短期记忆，需要在创建智能体时指明一个 checkpointer

• 实例化 checkpointer
• 创建 agent 时传入 checkpointer 实例
• 调用 agent 时，指定一个自定义的 thread_id

from langgraph.checkpoint.memory import InMemorySaver  # 基于内存的 checkpointer

agent = create_agent(
    'gpt-5', tools=[get_user_info],
    # 指定 checkpointer
    checkpointer=InMemorySaver()
)
from langchain_core.runnables import RunnableConfig
# 配置项中指明 checkpointer 组的 thread_id
config: RunnableConfig = {'configurable': {'thread_id': 'thread_1'}}
resp1 = agent.invoke(
    {'message': [HumanMessage(content='衡山在哪里')]},
    config
)
resp2 = agent.invoke(
    {'message': [HumanMessage(content='我上次问的是哪座山的位置')]},
    # 同一个 thread_id
    config
)
# 
print(resp2)

langchain 会自行将 checkpointer 中记录的消息进行拼接，组成消息列表，发送给大模型

基于数据库的 checkpointer 需要安装依赖。根据文档open in new window安装不同数据库所需的依赖

这里以 postgresql 与 sqlite 为例

• postgresql

  uv add langgraph-checkpoint-postgres
  

  from langgraph.checkpoint.postgres import PostgresSaver
  
  DB_URL = 'postgresql://postgres.postgres@127.0.0.1:5442/postgres?sslmode=disable'
  # 实例化时使用的是上下文管理器实现数据库的连接管理
  with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
      checkpointer.setup() # 自动创建数据表
      agent = create_agent(
          "gpt-5.5",
          tools=[get_user_info],
          checkpointer=checkpointer,
      )
  

• sqlite

  uv add langgraph-checkpoint-sqlite
  

  from langgraph.checkpoint.sqlite import SqliteSaver
  # 创建连接实例     将会在本地文件夹内创建 resources/checkpint.db   关闭检查（sqlite 会检查创建连接的线程与后续执行 sql 使用的线程是否相同）
  connection = sqlite3.connect('resources/checkpint.db', check_same_thread=False)
  checkpointer = SqliteSaver(connection)
  checkpointer.setup() # 自动创建数据表
  agent = create_agent(
      "gpt-5.5",
      tools=[get_user_info],
      checkpointer=checkpointer,
  )
  

# 记忆策略管理

由于随着会话推进，消息越来越多。而大模型一次接收的上下文有限，如果超出大小，则出现截取，即记得后面的，忘记前面的。

langchain 提供了记忆管理的策略open in new window来解决这个问题。

• 修剪（Trim）：移除消息数组中的前 N 或后 N 条消息，再进行发送。会有记忆丢失
• 删除（Delete）：删除 AgentState 快照。即该会话的历史消息直接清除
• 总结摘要（Summarize）：将早期的历史消息进行总结，得到消息摘要，替换消息数组中被总结的消息。需要使用 LLM 做总结摘要模型

  from langchain.agents.middleware import SummarizationMiddleware  # 总结摘要中间件
  
  agent = create_agent(
      model='gpt-4.1', 
      # 配置智能体的中间件
      middleware=[
          # 消息摘要中间件的实例化
          SummarizaionMiddleware(
              model='gpt-4.1-mini',  # 总结摘要模型 
              trigger=('tokens', 4000),  # 触发总结的方式: fraction - 达到模型最大 token 的百分比 / tokens - 按 token 数 / messages - 按消息输 
              keep=('messages', 20)  # 保留多少条消息
          )
      ])
  
• 自定义（Custom）：用户自定义策略，比如对消息进行筛选后传递

# 中间件

中间件实现了拦截。总结摘要中间件 SummarizationMiddleware 本质是将输入的提示词进行拦截，获取传入的历史消息数组，检测是否需要进行压缩，并根据初始化时的参数对历史消息数组进行压缩。

除了 langchain 自带的中间件，也可以自定义中间件。使用内置的装饰器 langchain.agent.middleware 定义中间件，不同拦截时机的装饰器分类如下

• 节点式钩子（执行点顺序拦截）
  • @before_agent: agent 执行之前拦截
  • @after_agent: agent 执行之后拦截
  • @before_model: model 执行前拦截
  • @after_model: model 执行后拦截
• 针对工具和模型的包装式钩子(替换对应的步骤)
  • @wrap_model_call: 每个模型调用时拦截
  • @wrap_tool_call: 每个工具调用时拦截

以下是自定义中间件示例，注意参数

from langchain.agent.middleware import before_agent, wrap_model_call, wrap_tool_call
from langchain.agents import AgentState
from langchain.runtime import Runtime

# 节点式钩子 的参数固定为 state: AgentState, runtime: Runtime, 返回值为空
@before_agent
def log_before_agent(state: AgentState, runtime: Runtime) -> None:
    # 在 agent 执行前自动调用该函数，并传入 state 与 runtime
    # runtime.context 就是上下文，是一个 dict
    print(f"要开始调用智能体了，附带的消息个数是{len(state['messages'])}")  # state['messages'] 即为传递给智能体的历史消息数组

from langchain.tools.tool_node import ToolCallRequest
from langchain_core.messages import ToolMessage
from langgraph.types import Command

# 包装式钩子 的参数固定为 request(模型调用的请求), handle(处理对象), 需要手动指定让处理函数处理请求。
@wrap_model_call
def model_call_hook(request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command]) -> ToolMessage | Command:
    print('模型在调用')
    # request.runtime 就是 Runtime
    return handler(request)  # 模型调用的过程会被替换成该函数的内容，返回处理请求结果，使模型正常调用

@wrap_tool_call
def tool_call_hook(request, handler):
    print(f"调用了工具: {request.tool_call['name']}, 传入的参数是 {request.tool_call['args']}")
    return handler(request)

根据日志顺序可知，执行顺序如下，可选内容是调用工具的部分

before_agent-> before_model -> model_call_hook -> after_model [-> tool_call_hook -> before_model -> model_call_hook -> after_model] -> after_agent

除了以上中间件外，生成提示词前也可以进行注入。可通过流程判断，实现提示词的选择

from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def report_prompt_switch(request: ModelRequest):
    # 在满足某个特定条件，比如调用了某个工具时。可通过 before_model 的钩子向上下文 dict 中注入一个变量，此处在获取该变量，进而方便后续选择提示词
    is_some_situation = request.runtime.context.get('flag')
    if is_some_situation:  # 当是某种特定条件时，返回对应的提示词
        return PromptTemplate(...)  
    return PromptTemplate(...)  # 不是特定场景，反正正常的提示词

# 智能体案例

AI 私厨管家，用户上传食材的照片，管家识别图片中的食材，根据食材搜索相关食谱推荐给用户

• 流程分析：图片识别 -> 智能搜索 -> 智能排序 -> 创意建议（找不到合适食谱时，提供创意搭配建议） -> 对话交互

• 定义模型

  from langchain.chat_models import init_chat_model
  model = init_chat_model(
      model='qwen3.5-plus',
      model_provider='openai',
      base_url=os.getenv('DASHSCOPE_BASE_URL'),
      api_key=os.getenv('DASHSCOPE_API_KEY')
  )
  

• 定义工具

  from langchain_tavily import TavilySearch
  # web搜索工具，使用 tavily 作为 web 搜索工具
  web_search = TavilySearch(max_results=5, topic='general')
  

• 记忆管理

  from langgraph.checkpoint.sqlite import SqliteSaver
  # 本地文件夹内创建 resources/checkpint.db             关闭检查（sqlite 会检查创建连接的线程与后续执行 sql 使用的线程是否相同）
  connection = sqlite3.connect('resources/personal_chief.db', check_same_thread=False)
  checkpointer = SqliteSaver(connection)
  checkpointer.setup() # 自动创建数据表
  

• 定义智能体：最关键的是系统提示词，需要强调优先使用工具

  from langchain.agents import create_agent
  system_prompt = """你是一名私人厨师。收到用户提供的食材照片或清单后，按照以下流程操作：
  1.识别和评估食材：若用户提供照片，首先辨别所有可见食材。基于食材的外观状态，评估其新鲜度与可用量，整理出一份“当前可用食材清单”
  2.智能食谱检索：优先调用 web_search 工具，以“可用食材清单”为核心关键词，查找可行菜谱
  3.多维度评估与排序：从营养价值和制作难度两个维度对检索到的候选食谱进行量化打分，并根据得分排序，制作简单且营养丰富的排名靠前
  4.结构化方案输出：把排序后的食谱整理为一份结构清晰的建议报告，要包括食谱信息、得分、推荐理由、食谱的参考图片，帮助用户快速做出决策
  请严格按照流程，优先调用 web_search 工具搜索食谱，搜索不到的情况下才能自己发挥
  """
  agent = create_agent(model=model, tools=[web_search], system_prompt=system_prompt, checkpointer=checkpointer)
  

• 测试：测试当前智能体是否可用

  from langchain.messages import HumanMessage
  # 创建一个多模态消息进行测试，图片使用网上搜索的图片
  multmodal_message = HumanMessage([
      {'type': 'text', 'text': '帮我看看能做什么'}
      {'type': 'image', 'url': 'https://xxx'}
  ])
  config = {'configurable': {'thread_id': '1'}}
  response = agent.invoke({'messages': [multimodal_message]}, config)
  for message in reponsep['messages']:
      message.pretty_print()
  response = agent.invoke({'messages': [HumanMessage('我喜欢第2道菜，详细说一说')]})
  print(response['messages'][-1].pretty_print())
  

• 基于 langsmith 的本地部署。langsmith 是一个在线平台，需要注册并生成 API_KEY。注意，langsmith 部署生成环境需要收费，该部署方法常用于本地测试服务进行调试

  • 配置环境变量

    LANGSMITH_API_KEY=xxxxx
    # 开启 langsmith 追踪
    LANGSMITH_TRACING_V2=true
    # 自定义项目名
    LANGSMITH_PROJECT=personal-chief
    
  • 将创建智能体代码集中到单个 py 文件。注意以下几点
    • 不需要会话记忆的 checkpointer 代码，langgraph 部署时会自动提供 checkpointer
    • 不需要调用智能体，只需创建好 agent。langgraph 自带 restful API 接口

    import os
    from dotenv import load_dotenv
    from langchain.chat_models import init_chat_model
    from langchain_tavily import TavilySearch
    from langchain.agent import create_agent
    
    load_dotenv()
    model = init_chat_model(
        model='qwen3.5-plus',
        model_provider='openai',
        base_url=os.getenv('DASHSCOPE_BASE_URL'),
        api_key=os.getenv('DASHSCOPE_API_KEY')
    )
    web_search = TavilySearch(max_results=5, topic='general')
    system_prompt = """你是一名私人厨师。收到用户提供的食材照片或清单后，按照以下流程操作：
    1.识别和评估食材：若用户提供照片，首先辨别所有可见食材。基于食材的外观状态，评估其新鲜度与可用量，整理出一份“当前可用食材清单”
    2.智能食谱检索：优先调用 web_search 工具，以“可用食材清单”为核心关键词，查找可行菜谱
    3.多维度评估与排序：从营养价值和制作难度两个维度对检索到的候选食谱进行量化打分，并根据得分排序，制作简单且营养丰富的排名靠前
    4.结构化方案输出：把排序后的食谱整理为一份结构清晰的建议报告，要包括食谱信息、得分、推荐理由、食谱的参考图片，帮助用户快速做出决策
    请严格按照流程，优先调用 web_search 工具搜索食谱，搜索不到的情况下才能自己发挥
    """
    agent = create_agent(model=model, tools=[web_search], system_prompt=system_prompt)
    
  • 使用 langgraph 本地部署测试服务。
    • 安装依赖：uv add langgraph-cli[inmem]
    • 创建配置文件 langgraph.json

      {
          "dependencies": ["."],  // 依赖项位置
          "graphs": {
              // 智能体实例的路径
              "chief_agent": "./personal_chief.py:agent",
          },
          "env": ".env"
      }
      
    • 启动 agent：uv run langgraph dev。给出三个链接
      • API: API地址
      • Studio UI: 调试页面
      • API Docs: API 文档
  • langsmith 的追踪数据获取。每月 5000 次免费追踪额度

    from langsmith import Client
    from langchain_core.tracers.context import collect_runs
    
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = ChatPromptTemplate.from_template("请用一句话解释什么是 {topic}")
    chain = prompt | llm
    
    # 1. 使用 collect_runs 实时监听链路
    with collect_runs() as cb:
        response = chain.invoke({"topic": "量子力学"})
        # 拿到本次调用的根 Run ID
        run_id = cb.traced_runs[0].id
    
    print(f"模型输出结果: {response.content}")
    print(f"本次追踪的 Run ID: {run_id}\n")
    
    # 2. 通过 LangSmith Client 在本地拉取该 Run 的完整追踪详情
    client = Client()
    run_details = client.read_run(run_id, load_child_runs=True)
    
    # 3. 提取本地需要记录的核心字段
    trace_data = {
        "run_id": str(run_details.id),
        "name": run_details.name,
        "inputs": run_details.inputs,           # 最终传递给链/模型的输入
        "outputs": run_details.outputs,         # 最终输出结果
        "total_tokens": run_details.total_tokens, # 消耗的总 Token 数
        "latency": run_details.latency,         # 耗时（秒）
        "start_time": str(run_details.start_time),
        "child_steps": [                        # 链路的子步骤（例如 Prompt 渲染、LLM 调用）
            {
                "step_name": child.name,
                "inputs": child.inputs,
                "outputs": child.outputs
            }
            for child in (run_details.child_runs or [])
        ]
    }
    

• 生产环境部署，使用 web 服务。

  • 图片等静态资源使用 oss 服务进行存储，发送给模型图片链接而不是 b64 编码，减少 token 消耗。此处以阿里 OSS 为例 uv add alibabacloud-oss-v2

    import alibabacloud_oss_v2 as oss
    from fastapi import APIRouter
    from datetime import timedelta
    # 路由实例
    router = APIRouter()
    # oss 身份认证 - 需要三个环境变量 OSS_ACCESS_KEY_ID OSS_ACCESS_KEY_SECRET OSS_BUCKET
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    # 加载默认配置
    cfg = oss.config.load_default()
    # 设置身份认证
    cfg.credentials_provider = credentials_provider
    # 添加地区，会根据地区自动构建访问域名
    cfg.region = 'cn-beijing'
    # 使用配置对象创建 OSS 实例
    client = oss.Client(cfg)
    # 配置信息
    OSS_ENDPOINT = os.getenv('OSS_ENDPOINT', 'oss-cn-beijing.aliyuncs.com')
    OSS_BUCKET = os.getenv('OSS_BUCKET')
    # 获取 文件上传的授权签名 接口
    @router.get('oss/presign')
    def chat_endpoint(filename: str):
        # 扩展名获取 Content-Type
        content_type_map = {
            'jpg': 'image/jpeg',
            'jpeg': 'image/jpeg',
            'png': 'image/png',
            'gif': 'image/gif',
            'webp': 'image/webp'
        }
        ext = filename.split('.')[-1].lower() if '.' in filename else 'jpg'
        content_type = content_type_map.get(ext, 'application/octet-stream')
        pre_result = client.presign(oss.PutObjectRequest(
            bucket=OSS_BUCKET,
            key=filename,
            content_type=content_type,
        ), expires=timedelta(seconds=3600))
        # 返回上传 URL 和可访问的图片路径 - 前端通过该接口获取的带签名 URL 自行上传到 OSS
        return {
            'uploadUrl': pre_result.url.strip('"'),
            'contentType': content_type,
            'accessUrl': f'https://{OSS_BUCKET}.{OSS_ENDPOINT}/{filename}'
        }
    

  • 主程序

    app = FastAPI(title='Presonal Chief', description='私厨', version='0.1.0')
    # 跨域
    app.add_middleware(CORSMiddleware, allow_origins=['*'], allow_credentials=True, allow_methods=['*'], allow_headers=['*'])
    # 路由
    app.include_router(chat.router, prefix='/api', tags=['对话'])
    app.include_router(oss.router, prefix='/api', tags=['申请上传签名url'])
    # 静态资源挂载
    static_dir = os.path.join(os.path.dirname(__name__), 'static')
    if os.path.exists(static_dir):
        app.mount('/', StaticFiles(directory=static_dir, html=True), name='static')
    # 静态资源接口。路由处理，判断是后端服务还是静态资源
    @app.get('/{path:path}', include_in_schema=False)
    async def serve_frontend(path: str):
        # 排除 API 路径
        if path.startswith('api/'):
            from fastapi.responses import JSONResponse
            return JSONResponse({'error': 'Not Found'}, status_code=404)
        # 静态资源，直接返回
        file_path = os.path.join(static_dir, path)
        # 找不到对应页面，返回主页
        if not os.path.isfile(file_path):
            file_path = os.path.join(static_dir, 'index.html')
        if os.path.exists(file_path):
            return FileResponse(index_path)
        return {'message': '你的独家私厨上线了~', 'status': 'ok'}
    
    if __name__ == '__main__':
        import uvicorn
        # app/main.py
        # python -m app.main
        uvicorn.run('app.main:app', host='127.0.0.1', port=8001, reload=True)
    

  • agent 模块：需要添加记忆管理功能

    # agents/personal_chief.py
    import os
    from dotenv import load_dotenv
    from langchain.chat_models import init_chat_model
    from langchain_tavily import TavilySearch
    from langchain.agent import create_agent
    from langgraph.checkpoint.sqlite import SqliteSaver
    load_dotenv()
    model = init_chat_model(
        model='qwen3.5-plus',
        model_provider='openai',
        base_url=os.getenv('DASHSCOPE_BASE_URL'),
        api_key=os.getenv('DASHSCOPE_API_KEY')
    )
    web_search = TavilySearch(max_results=5, topic='general')
    connection = sqlite3.connect('../db/personal_chief.db', check_same_thread=False)
    checkpointer = SqliteSaver(connection)
    checkpointer.setup() # 自动创建数据表
    system_prompt = """你是一名私人厨师。收到用户提供的食材照片或清单后，按照以下流程操作：
    1.识别和评估食材：若用户提供照片，首先辨别所有可见食材。基于食材的外观状态，评估其新鲜度与可用量，整理出一份“当前可用食材清单”
    2.智能食谱检索：优先调用 web_search 工具，以“可用食材清单”为核心关键词，查找可行菜谱
    3.多维度评估与排序：从营养价值和制作难度两个维度对检索到的候选食谱进行量化打分，并根据得分排序，制作简单且营养丰富的排名靠前
    4.结构化方案输出：把排序后的食谱整理为一份结构清晰的建议报告，要包括食谱信息、得分、推荐理由、食谱的参考图片，帮助用户快速做出决策
    请严格按照流程，优先调用 web_search 工具搜索食谱，搜索不到的情况下才能自己发挥
    """
    agent = create_agent(model=model, tools=[web_search], checkpointer=checkpointer, system_prompt=system_prompt)
    

  • 聊天接口

    • 流式对话接口

      # models/schemas.py
      class ChatRequest(BaseModel):
          message: str
          image_url: Optional[str] = None
          thread_id: str
      # agents/personal_chief.py   放在 agent 声明的文件中，因为用到了声明时的内容
      from langchain.messages import AIMessageChunk
      async def searcb_recipes(prompt: str, image: str, thread_id: str):
          """查询 agent 搜索食谱"""
          logger.info(f'[user]: {prompt}, image: {image}, thread_id: {thread_id}')
          try:
              if not image or image.strip() == '':
                  message = HumanMessage(prompt)
              else:
                  message = HumanMessage(content=[
                      {'type': 'image', 'url': image},
                      {'type': 'text', 'text': prompt}
                  ])
              for chunk, metadata in agent.stream(
                  {'messages': [messaege]}, {'configurable': {'thread_id': thread_id}}, stream_mode='messages'
              ):
                  # 判断是 AI 响应的消息，而不是调用工具等其他消息，再进行返回
                  if isinstance(chunk, AIMessageChunk) and chunk.content:
                      yield chunk.content
          except Exception as e:
              logger.exception(e)
              yield '信息检索失败，试试看手动输入食物列表？'
      # api/chat.py
      from fastapi.responses import StreamingResponse
      @router.post('chat/stream')
      async def chat_endpoint(request: ChatRequest):
          retutn StreamingResponse(
              # yield 函数
              searcb_recipes(request.message, request.image_url, request.thread_id)
              # mimetype
              media_type='text/event-stream'
          )
          
      
    • 获取历史消息接口

      # checkpointer.get({'configurable': {'thread_id': '1'}}) - 一个大 dict，根据对应 key 取出其中的消息内容
      # for m in checkpointer.get(config)['channel_values']['messagess']
      def get_messages(thread_id: str) -> list[dict[str, str]]:
          checkpoint = checkpointer.get({'configurable': {'thread_id': thread_id}})
          if not checkpointer: return []
          channel_values = checkpoint.get('channel_values')
          if not channel_values: return []
          messages = channel_values.get('messagess')
          if not messages: return []
          # 只保留用户与 AI 之前的消息，过滤工具调用等消息
          result = []
          for msg in messages:
              if not msg.content: continue
              if isinstance(msg, HumanMessage):
                  result.append({'role': 'user', 'content': msg.content})
              elif isinstance(msg, AIMessage):
                  result.append({'role': 'assistant', 'content': msg.content})
          return result
      
      @router.get('chat/messages')
      async def get_chat_messages(thread_id: str):
          return {'messages': get_messages(thread_id)}
      
    • 清空历史消息接口

      @router.delete('chat/messages')
      async def clear_chat_messages(thread_id: str):
          # agent 声明时的实例
          # checkpointer.delete_thread('<thread_id>')
          clear_messages(thread_id)
          return {'success': True}
      

# RAG

检索增强生成（Retrieve-Augmented Generation），为大模型提供了从特定数据源检索到的信息，以此来修正和补充生成的答案

核心流程

• 离线处理：分片 + 索引
• 在线处理：召回 + 重排 + 生成

# 余弦相似度

即一个向量在另一个向量上投影的大小，越大表明相似度越大

• 计算点积

  def get_dot(vec_a, vec_b):
      if len(vec_a) != len(vec_b):
          raise ValueError('2个向量维度必须相同')
      dot_sum = 0
      for a, b in zip(vec_a, vec_b):
          dot_sum += a * b
      return dot_sum
  
• 计算向量长度

  def get_norm(vec):
      sum_square = 0
      for v in vec:
          sum_square += v * v
      return numpy.sqrt(sum_square)
  
• 计算余弦相似度

  def cosine_similarity(vec_a, vec_b):
      return get_dot(vec_a, vec_b) / (get_norm(vec_a) * get_norm(vec_b))
  

# 文档加载器

langchain 内置了文档加载器，用于将不同来源的数据读取为文档格式

最终被加载为 from langchain_core.documents import Documents 的实例。返回 list[Document] 或 Iterator[document]

# 每条数据被解析为一个 Document，元数据可能包含来源的文件路径，该数据所在文档中的位置
Document(page_content="文本内容", metadata={"source": "file.csv", ...})

文档加载器都实现了 load 与 lazy_load 方法，用于加载数据。其中 lazy_load 用于流式加载，避免内存溢出

from langchain_community.document_loaders import CSVLoader, JSONLoader, TextLoader

csv_loader = CSVLoader(...)
# 一次性加载整个文件内容，得到 list[Document]
documents = loader.load()
# 文件内容分段加载，得到 Iterator[document]。大文件常用此加载避免内存溢出
for document in loader.lazy_load():
    # 一个一个文档进行处理
    print(document)

• CSVLoader

  loader = CSVLoader(
      file_path='./xxx.csv',
      # 以下为可选参数
      csv_args={
          'delimiter': ',',  # 指明 csv 使用的分隔符
          'quotechar': '"',  # 指明 csv 中转义字符使用的引号。csv 的数据字符串中如果包含分隔符，会用引号包裹整个字符串实现转义
          'fieldnames': ['name', 'age', 'gender']  # 表头，不设定默认使用第一行作为表头。当数据没有表头时设定该行，否则不使用
      },
      encoding='utf-8'  # 指定读取时采用的编码，win 默认会使用 gbk 编码读取文件内容，即使文件是 utf-8
      source_column=''  # 指定数据来源列的列名。比如表格中存在一列 platform，值分别为 b站，小红薯，微博，此时可传入 `platform`，Document 的元数据中将存在 source='b站' 的内容
  )
  data = loader.load()
  print(data)  
  # 得到每行一个 Document 对象组成的 list[Document]
  # Document.page_content 就是抽取的内容，以 3 个表头为例，抽取出的 page_content 字符串如下
  # header1: value1\nheader2: value2\nheader3: value3
  

• JSONLoader: 需要额外安装依赖 jq，一个跨平台 json 解析工具

  • jq 语法
    • 最开始的 . 表示根路径
    • .name 获取最外层（根路径）的 key 为 name 的值
    • 对应的值为数组时，使用 [] 获取指定索引的内容
    • .addr.city 多层嵌套时使用 . 进行内层访问
    • 对应的值为 list[dict] 时
      • 使用 []. 获取 list 内的全部 dict。此时末尾的 . 表示 dict 本身
      • 使用 [].key1 获取 list 内的全部 dict 中 key1 对应的值
  • JSONLoader 的使用，在创建实例时就需要明确 jq 抽取语法，指定抽取的内容。始终返回 list[Document]，只抽取一个时数组内只有一个元素

    loader = JSONLoader(
        file_path='xx.json',  # 文件路径
        jq_schema='.[].',  # jq 的抽取语法。
        text_content=False,  # 抽取的是否为字符串，默认为 True。如果是想抽取出 dict 的情况，比如 list[dict] 使用 `.[].`，需要设置为 False
        json_lines=True  # 文件是否为 JsonLines 文件，即每一行都是 json 的文件（外层没有 list 括号）
    )
    # Tips:
    # 1. 若抽取出的是 dict，则 Document.page_content 是 json dumps 字符串，需要使用 json.loads 转换成 dict
    # 2. 若 json_lines 文件，语法无需按照 list[dict] 使用 `.[]` 处理，直接按单个 json 处理
    #    eg: jsonlines文件的每一行都是{'name': 'xxx'}, 此时抽取 name 值无需 `.[].name`，直接 `.name`
    

• TextLoader: 加载 txt 文件中的数据，将全部内容放入一个 Document 对象中，返回长度为 1 的 list[Document]

  loader = TextLoader(file_path='xxx.txt', encoding='utf-8')
  

• PyPDFLoader: 该加载器依赖 PyPDF，因此需要额外安装依赖 pypdf

  loader = PyPDFLoader(
      file_path='',
      mode='page',  # Document切分模式，默认为 `page` - 按页转换 Document，或 `single` - 文件作为一个 Document
      password='',  # 文件密码
  )
  

# 文档分割器

对于一些内容长度超大的文档，比如 TextLoader 加载内容较多的 txt 文件，官方提供了 RecursiveCharacterTextSplitter 递归字符串文本分割器，用于按自然段落分割大文档

保持了上下文完整性和控制片段大小之间实现了良好平衡，开箱即用效果佳

如果没有，手动安装依赖 langchain_text_splitters

from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader(...)
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # 每段文本的最大数量
    chunk_overlap=50,  # 每段之间重叠的字符数
    separators=['\n\n', '\n', '。', '！', '？', '.', '!', '?', ' ', ''],  # 文本分段依据
    length_function=len  # 统计文本长度的函数
)
# 分割 document 对象
split_docs = splitter.split_documents(docs)
# 分割 str
split_strs = splitter.split_text(text)

# 向量模型

向量模型对象拥有 embed_query 方法与 embed_documents 方法，一个用于转换 str，一个用于转换 list[Document]

• openai

  from langchain_openai import OpenAIEmbeddings
  # 实例化
  embed = OpenAIEmbeddings()
  embed.embed_query('转换一个文本信息')
  embed.embed_documents(['转换多个文本信息1', '转换多个文本信息2'])
  
• 千问

  from langchain_community.embeddings import DashScopeEmbeddings
  
  # 实例化，默认使用 text-embedding-v1
  embed = DashScopeEmbeddings()
  embed.embed_query('转换一个文本信息')
  embed.embed_documents(['转换多个文本信息1', '转换多个文本信息2'])
  
• ollama

  from langchain_ollama import OllamaEmbeddings
  embed = OllamaEmbeddings(model='qwen3-embedding:4b')  # 传入 ollama 中启动的模型
  embed.embed_query('转换一个文本信息')
  embed.embed_documents(['转换多个文本信息1', '转换多个文本信息2'])
  

# 向量存储

需要声明向量存储对象（VectorStore类）实现存储，查询，删除等功能。该对象含有以下常用方法

• add_documents 方法，传入文档列表，内部实现文档转换成向量，存储 id | 向量 | 原文 的一一对应数据

• add_texts 方法，传入文档原文 list[str]，内部实现文本转成向量，存储 id | 向量 | 原文 的一一对应数据。该方式常用于测试

• delete 方法，传入要删除的 id 列表 list[str] 实现对指定文档的删除

• similarity_search 方法查询向量库，传入问题字符串与召回数量，内部实现问题字符串转换成向量，并查询数据库，返回匹配度最高的指定数量的文档结果 list[Document]

搜索相似度高的结果

• 内置向量存储。采用内存存储向量，主要做测试使用

  from langchain_core.vectorstores import InMemoryVectorStore
  from langchain_openai import OpenAIEmbeddings
  # 实例化向量存储对象，传入向量模型
  vector_store = InMemoryVectorStore(OpenAIEmbeddings())
  
  # 手动传入 list[Document] 与 手动指定的IDs list[str]（可选）。内部实现文档转换成向量，存储 id | 向量 | 原文 的一一对应数据
  vector_store.add_documents(documents=[...], ids=[...])
  # 查询，传入问题字符串与召回数量
  results: list[Document] = vector_store.similarity_search("查询问题", k=3)
  
• 外部向量存储。采用向量数据库存储数据，此处以轻量数据库 chroma 为例。需要安装依赖 langchain-chroma 与 chromadb

  from langchain_chroma import Chroma
  from langchain_community.embeddings import DashScopeEmbeddings
  
  vector_store = Chroma(
      collection_name='example_collection',  # 指明集合
      embedding_function=DashScopeEmbeddings(),  # 指明向量模型
      persist_directory='./example.db'  # 文件存储路径
  )
  

chroma 类似 sqlite，会生成一个 db 文件存储数据

生产环境可考虑使用 milvus 进行向量存储与查询

# 向量查询

• similarity_search 方法支持 filter 参数传入 dict[str, str] 进行元数据过滤。比如 vector_store.similarity_search('xxx', 3, filter={'source': 'b站'}) 进行元数据过滤。最终将只会在满足条件的 Document 里
• 构建提示词

  from langchain_core.prompts import ChatPromptTemplate
  from langchain_core.vectorstores import InMemoryVectorStore
  from langchain_community.chat_models import ChatTongyi
  from langchain_community.embeddings import DashScopeEmbeddings
  model = ChatTongyi(model='qwen3-max')
  prompt = ChatPromptTemplate.from_messages([
      ('system', '你是一位知识助手，请根据已知参考材料回答用户的问题，生成准确的答案。相关片段: {context}'),
      ('user', '用户问题：{input}')
  ])
  vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model='text-embedding-v4'))  # 手动指明使用 v4 模型
  vector_store.add_texts([..., ..., ...])
  input_text = ''  # 用户的提问
  # 召回，即使转换时用的是 list[str]，查询出来依然返回 list[Document]，只是元数据都是空 dict
  result: list[Document] = vector_store.similarity_search("查询问题", k=3)
  # 组装参考文本
  reference_text = '\n- '.join([doc.page_content for doc in result])
  # 调试提示词函数
  def print_prompt(prompt):
      print(prompt.to_string())
      print('='*20)
      return prompt
  # chain
  chain = prompt | print_prompt | model | StrOutputParser()
  # 传入提示词变量，调用模型
  print(chain.invoke({'input': input_text, 'context': reference_text}))
  

# RunnablePassthrough

根据以上流程，向量检索的召回步骤是单独提炼出来，在 chain 外运行的。这是由于向量存储对象不是 Runnable 的子类，无法放入链中

此时可使用 RunnablePassthrough 类将检索流程入链

• 场景一：透传输入（保持原始值不变）。当你的下游步骤需要原始输入，而并行分支需要处理该输入时，使用 RunnablePassthrough() 透传

  # 经典的 RAG 检索链
  chain = ({
      "context": retriever,               # 分支 1：拿着 question 去检索出文档
      "question": RunnablePassthrough()   # 分支 2：把输入的 question 原封不动透传给 Prompt。本质就是个占位符，会替换为上一步传入的内容
  } | prompt | model | StrOutputParser())
  
  chain.invoke("什么是 LangGraph？")
  
• 场景二：RunnablePassthrough.assign()（在字典中动态追加字段）。如果你的输入本身就是一个字典，你只想在保留原有所有 key 的同时，动态追加一些新的衍生字段，assign() 是最优雅的标准方案

  chain = (
      RunnablePassthrough.assign(
          context=lambda x: retriever.invoke(x["question"])  # 追加 context 字段，保留原有的 question
      ) | prompt | model)
  

在 langchain 中的向量存储对象拥有 as_retriever 方法，可以返回 Runnable 子类的实例对象，该实例对象由于是 Runnable 子类，可直接使用 invoke / stream 方法进行相似度搜索

# 构建模型实例
...

vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model='text-embedding-v4'))
# 正常做索引这种准备工作
vector_store.add_texts(...)
# 获取向量存储的 Runnable 子类实例对象，用于后续入链
retriever = vector_store.as_retriever(search_kwargs={'k': 2})  # 搜索参数中的召回数量通常是固定值，可提前将该搜索参数传入

input_text = ''  # 用户问题

# 构建 chain，此时要考虑每个环节的输入与输出
#   - retriever: i-用户问题，o-检索结果 list[Document]
#   - prompt: i-用户问题与检索结果 dict，o-完整提示词
#   因此不能使用 chain = retriever | prompt，因为 retriever 的输出与 prompt 的输入内容与类型都不匹配。prompt 不仅需要召回结果，还需要用户原始问题的透传
# 由于 dict 是 Mapping，查看 Runnable 发现可以入链，为了保存用户输入与召回结果，使用 dict 入链
from langchain_core.runnables import RunnablePassthrough
chain = (
    {
        # 用户问题的透传，即上一步传入内容(用户问题)将替换掉 RunnablePassthrough()
        'input': RunnablePassthrough(), 
        # 内部再构造一个子 chain，将上一步的传入(用户问题)入 chain，经过向量存储的查询，到输出内容格式化函数，得到纯文本内容，用于后续注入提示词模版
        'context': retriever | docs_format
    } ｜ prompt | print_prompt | model | StrOutputParser()
)
# 将召回的 list[Document] 转换成字符串，用于后续注入提示词模版
def docs_format(docs: list[Document]) -> str:
    if not docs: return '无相关参考资料'
    return '\n- '.join([doc.page_content for doc in docs])

print(chain.invoke(input_text))

# 案例

# 服装商品的知识库

可用户提供身高体重等数据，提供相关的尺码推荐。也可提供指定产品的保养方法等内容。文件结构如下

• chat_history/: 对话历史存储目录
• chroma_db/: 数据库文件目录
• data/: 参考文档目录
• app_file_upload.py: 知识库更新主程序（streamlit前端框架）。向数据库中上传参考文档
• app_qa.py: 项目主程序，web 服务
• config_data: 配置文件
• file_history_store.py: 会话存储
• knowledge_base.py: 知识库更新服务。将上传的参考文档解析入库
• rag.py: RAG 服务
• vector_stores.py: 向量存储服务

# 离线流程

分片与索引阶段。用户上传文件，进行文本去重，解析成向量，存入数据库

上传基于 streamlit 框架，需要安装依赖 streamlit。该框架可构建前端页面，使用 streamlit run <py文件>

• 获取文件上传内容 app_file_upload.py streamlit 编写上传文件的静态页面

  import streamlit as st
  # 添加一个 h1
  st.title('知识库更新服务')
  # 文件上传
  uploader = st.file_uploader(
      '请上传 TXT 文件',  # 文件上传元素的 label
      type=['txt'],  # 支持上传文件的类型
      accept_multiple_files=False
  )
  # 获取上传文件的内容
  if uploader is not None:
      file_name = uploader.name
      file_type = uploader.type
      file_size = uploader.size / 1024  # kb
      st.subheader(f'文件名：{file_name}')
      st.write(f'格式：{file_type} | 大小：{file_size:.2f} KB')  # 保留2位小数
  
      # 获取文件内容
      text = uploader.get_value().decode('utf-8')
  
• 文本去重与向量解析和存储 knowledge_base.py

  from hashlib import md5
  from langchain_chroma import Chroma
  from langchain_community.embeddings import DashScopeEmbeddings
  from langchain_text_splitters import RecursiveCharacterTextSplitter
  
  def check_md5(md5_str: str) -> bool:
      """检测传入的 md5 是否已经被处理"""
      if not os.path.exists(config.MD5_PATH):  # 使用一个文件存储已处理的 md5
          open(config.MD5_PATH, 'w', encoding='utf8').close()  # 不存在则创建该文件
          return False
      with open(config.MD5_PATH, 'w', encoding='utf8') as f:
          for line in f.readlines():  # 检测文件中是否存在该 md5 值
              if md5_str == line.strip():  # strip 移除末尾的换行
                  return True
      return False
  
  def save_md5(md5_str: str):
      """将传入的 md5 标记为已处理"""  # 使用一个文件存储已处理的 md5
      with open(config.MD5_PATH, 'w', encoding='utf8') as f:
          f.write(md5_str + '\n')
  
  def get_string_md5(md5_str: str, encoding='utf-8') -> str:
      """计算传入内容的 md5"""
      return md5(str(_in).encode()).hexdigest()
  
  # 解析存储
  class KnowledgeBaseService:
      def __init__(self):
          self.chroma = Chroma(  # 向量数据库对象
              collection_name=config.COLLECTION_NAME,  # 数据库集合名
              embedding_function=DashScopeEmbeddings(model='text-embedding-v4'),
              persist_directory=config.PERSIST_DIR  # 数据库文件存放路径
          )
          self.spliter = RecursiveCharacterTextSplitter(  # 文本分割器对象
              chunk_size=config.CHUNK_SIZE,  # 文本段长度 1000
              chunk_overlap=config.CHUNK_OVERLAP,  # 段间重复字符数 100
              separators=config.CHUNK_SEPARATORS,  # 段落分割符，同之前
              length_function=len  # 长度统计使用的函数
          )
  
      def upload_by_str(self, data: str, filename: str) -> str:
          """
          将传入的字符串解析成向量，并存入数据库
          :param data: 待解析的文本
          :param filename: 文本来源的文件名，用于构造存储元数据
          :return: 处理结果的文本消息
          """
          # 检测原始文本内容是否已经被处理
          md5_str = get_string_md5(data)
          if check_md5(md5_str): return '[Pass] 该文本已处理'
          
          # 原始文本大于块的大小时，再进行分割
          if len(data) > config.CHUNK_SIZE:
              chunks = self.splitter.split_text(data)  # type: list[str]
          else:
              chunks = [data]
      
          # 构建原始文本的元数据
          metadata = {'source': filename, 'create_time': datatime.now().strftime('%Y-%m-%d %H:%M:%S'), 'operator': '此处可记录上传用户的ID'}
          
          # 存入数据与元数据
          self.chroma.add_texts(chunks, metadatas=[metadata for _ in chunks])
  
          # 标记处理完的数据
          save_md5(md5_str)
  
          return '[Success] 该文本已成功入库'
  
• 离线服务开发 app_file_upload.py
  • 状态存储。streamlit 的热加载，当页面发生变化时，整个代码会重新加载，计数器，状态标志位等全局变量会重新加载为初始化的值。因此计数器、状态标志位等内容需要 session 存储

    # st.session_state 用于存储状态。该变量是一个 dict
    # 这里以计数器为例
    if 'counter' not st.session_state:
        st.session_state['counter'] = 0
    
    counter += 1
    
  • 向量存储服务的实例对象作为全局变量，需要使用 session_state 存储，不然每次都会新建

    if 'service' not st.session_state:
        st.session_state['service'] = KnowledgeBaseService()
    
    uploader = st.file_uploader(
        '请上传 TXT 文件',  # 文件上传元素的 label
        type=['txt'],  # 支持上传文件的类型
        accept_multiple_files=False
    )
    # 获取上传文件的内容
    if uploader is not None:
        file_name = uploader.name
        file_type = uploader.type
        file_size = uploader.size / 1024  # kb
        st.subheader(f'文件名：{file_name}')
        st.write(f'格式：{file_type} | 大小：{file_size:.2f} KB')  # 保留2位小数
    
        # 获取文件内容
        text = uploader.get_value().decode('utf-8')  # type: str
    
        # 将上传内容通过向量存储服务实例对象，解析存储进数据库中
        with st.spinner('Loading...'):  # 加载动画。解析存储需要一定时间
            res = st.session_state['service'].upload_by_str(text, file_name)
            st.write(res)  # 在页面上显示存储结果
    

# 在线流程

• 向量存储服务 vector_stores.py 获取 vector_store 实例对象（向量检索器），用于链中查询向量库时使用

  from langchain_chroma import Chroma
  
  class VectorStorceService:
      def __init__(self, embedding_model)
          self.embedding = embedding_model  # 向量模型对象。后续获取 vector_store 实例对象时，可以顺便获取向量模型实例对象
          self.vector_store = Chroma(  # 向量数据库对象
              collection_name=config.COLLECTION_NAME,  # 数据库集合名
              embedding_function=DashScopeEmbeddings(model='text-embedding-v4'),
              persist_directory=config.PERSIST_DIR  # 数据库文件存放路径
          )
      
      def get_retriever(self)：
          """获取能入链的 vector_store 实例对象"""
          return self.vector_store.as_retriever(search_kwargs={'k': config.TOP_K})  # 提前传入每次搜索时的固定参数，召回数量
  

• RAG 核心代码 rag.py

  from langchain_core.documents import Document
  from langchain_core.prompts import ChatPromptTemplate
  from langchain_core,runnables import RunnablePassthrough
  from langchain_core.output_parsers import StrOutputParser
  from langchain_community.embeddings import DashScopeEmbeddings
  from langchain_community.chat_models.tongyi import ChatTongyi
  
  from vector_stores import VectorStoreService
  
  class RagService:
      def __init__(self):
          self.vector_service = VectorStoreService(
              embedding_model=DashScopeEmbeddings(model='text-embedding-v4')
          )
          self.prompt_template = ChatPromptTemplate.from_messages([
              ('system', '你是一位知识助手，请根据已知参考材料回答用户的问题，生成简洁准确的答案。相关片段: {context}'),
              ('user', '用户问题：{input}')
          ])
          self.chat_model = ChatTongyi(model='qwen3-max')
          self.chain = self.__get_chain()
  
      def __get_chain(self):
          retriever = self.vector_service.get_retriever()
  
          # 处理 vector_store 查询数据库返回的 list[Document]，使其可以被后续环节作为输入
          def format_document(docs: list[Document]) -> str:
              if not docs: return '无相关参考资料
              return [f'\n- {doc.page_content}' for doc in docs]  # 将召回内容拼接成带有一定格式的字符串
  
          chain = ({
              'input': RunnablePassthrough(),
              'context': retriever | format_document  # 该子链最终返回字符串，作为后续提示词模版的内容注入
          } | self.prompt_template | self.chat_model | StrOutputParser())
  
          return chain
      
  # 测试该类是否可以正常使用
  if __name__ == '__main__':
      print(RagService().chain.invoke('体重200斤买哪个尺码的衣服'))
  

• 会话历史记录

  • 使用 json 文件存储消息历史 list[dict] file_history_store.py

    from langchain_core.chat_history import BaseChatMessageHistory
    
    class FileChatMessageHisttory(BaseChatMessageHistory):
        # 同之前的实现，使用 json 文件存储消息历史 list[dict]
        ...
    
    # 编写根据 session_id 获取历史消息实例对象，用于创建历史消息链时传入
    def get_history(session_id: str) -> FileChatMessageHisttory:
        return FileChatMessageHisttory(session_id, './chat_history/')
    
  • 改写创建链函数，使其具有存储历史消息功能 rag.py

    from langchain_core.runnables import RunnableWithMessageHistory
    
    from .file_history_store import get_history
    
    def __get_chain(self):
        retriever = self.vector_service.get_retriever()
    
        def format_document(docs: list[Document]) -> str:
            if not docs: return '无相关参考资料
            return [f'\n- {doc.page_content}' for doc in docs]
    
        def format_for_retriever(input_dict: dict) -> str:
            return input_dict['input']  # 传入用户问题的占位符
    
        def get_prompt_input(input_dict: dict) -> list[dict]:  # 构造系统提示词的输入
            return {
                'input': input_dict['input']['input'],  # 先取出穿透字典，再从中取出对应内容
                'history': input_dict['input']['history'],  # 先取出穿透字典，再从中取出对应内容
                'context': input_dict['context']
            }
    
        chain = ({ # 由于是历史消息转换链，传入内容是 dict，而不是 str
            'input': RunnablePassthrough(),  # 此时穿透的内容是 {'input': '<用户问题>', 'history': [<消息数组>]}
            # 子链的第一项需要将 dict 处理成用户问题 str，向量存储对象才能进行相似度搜索
            'context': RunnableLambda(format_for_retriever) | retriever | format_document  
        } 
        | RunnableLambda(get_prompt_input) | self.prompt_template  # 由于提示词现在需要处理三个占位符，原本的 dict 已无法使用
        | self.chat_model | StrOutputParser())
    
        return RunnableWithMessageHistory(
            chain,
            get_history,
            input_messages_key='input',
            history_message_key='history'  # 历史消息占位符
        )
    
    # 由于需要历史消息占位符，因此修改系统提示词
    from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
    
    self.prompt_template = ChatPromptTemplate.from_messages([
        ('system', '你是一位知识助手，请根据已知参考材料回答用户的问题，生成简洁准确的答案。相关片段: {context}'),
        ('system', '并且给出如下对话历史记录'),
        MessagesPlaceholder('history'),  # 新增历史消息占位符，并添加提示词说明这个内容是历史消息
        ('user', '用户问题：{input}')
    ])
    
    # 以下是测试代码
    if __name__ == '__main__':
        # 需要指明 session_id
        config = {'configurable': {'session_id': 'user_01'}}
        # 由于此时的 chain 是 RunnableWithMessageHistory 转换的链，输入内容必须是 {'<用户提示词占位符>': '<用户内容>'}
        # print(RagService().chain.invoke('体重200斤买哪个尺码的衣服'), config)
        print(RagService().chain.invoke({'input': '体重200斤买哪个尺码的衣服'}, config))
    

• 用户聊天前端页面 app_qa.py。由于是 python 代码，可直接调用编写的各个后端服务。streamlit run app_qa.py 进行调试

  import streamlit as st
  from rag import RagService
  
  st.title('智能客服')
  st.divider()  # 分割符
  
  # rag服务作为全局变量，需要 session_state 保存
  if 'rag' not in st.session_state:
      st.session_state['rag']= RagService()
  
  # prompt = st.chat_input()  # 聊天窗口与输入栏
  
  # 由于用户的输入 prompt 是全局变量，每次都会刷新整个页面，历史消息需要 session_state 保存
  if 'message' not in st.session_state:
      # 消息历史数组
      st.session_state['message'] = [{'role': 'assistant', 'content': '你好，请问有什么可以帮你'}]
  
  # 每次刷新都直接显示历史消息 - 此时内部使用了聊天窗口与打印，因此可将获取用户输入放到后面
  for message in st.session_state['message']:
      st.chat_message('role').write(message['content'])
  
  prompt = st.chat_input()
  if prompt:  # 获取用户的输入
      st.chat_message('user').write(prompt)  # 以用户身份显示获取的内容
      st.session_state['message'].append({'role': 'user', 'content': prompt})  # 向历史消息中追加
      
      ai_res_list = []
      with st.spinner('AI 思考中...'):
          res = st.session_state['rag'].chain.stream({'input': prompt}, config)
          
          # 捕获 yield 的内容，拼接得到消息全文，用于记录进数组，同时保持原始内容 yield 返回
          def capture(generator, cache_list):
              for chunk in generator：
                  cache_list.append(chunk)
                  yield chunk
  
          st.chat_message('assistant').write_stream(capture(res, ai_res_list))
  
          st,session_state['message'].append({'role': 'assistant', 'content': ''.join(ai_res_list)})  # 向历史消息中追加
  

# 智能商品客服

可处理售前与售后问题。包括售前的产品咨询，售后的操作指导、故障处理、维护建议等

代码open in new window

项目结构

• agent/
  • tools/
    • agent_tools.py 工具
    • middleware.py 中间件，用于提示词切换
  • react_agent.py 智能体
• config/ 配置信息 yaml 文件
• data/ rag参考资料、数据库文件路径
• prompts/ 提示词文件夹
• model/ 大模型
  • factory.py 大模型相关实例的创建
• utils/
  • logger_handler.py: 日志器
  • config_handler.py: rag、agent、chroma 等配置对象
  • file_handler.py: 文件相关操作。计算文件内容 md5、获取文件夹内的文件列表、pdf/txt加载器
  • path_tool.py: 项目路径。获取项目根目录函数与路径拼接函数
  • prompt_loader.py: 读取文件中的 系统提示词、rag提示词、报告生成提示词
• rag/
  • vector_store.py: 向量存储对象
  • rag_service.py: 总结服务

# rag 相关功能

需要实现向量存储服务，与rag核心逻辑

• 向量存储服务 vector_store.py 实现自动加载文档内容与相似度搜索。与之前代码类型

  class VectorStoreService:
      def __init__(self):
          self.vector_store = Chroma(  # 向量存储实例
              collection_name=chroma_conf["collection_name"],
              embedding_function=embed_model,
              persist_directory=get_abs_path(chroma_conf["persist_directory"]),
          )
          self.spliter = RecursiveCharacterTextSplitter(  # 文档分割器
              chunk_size=chroma_conf["chunk_size"],
              chunk_overlap=chroma_conf["chunk_overlap"],
              separators=chroma_conf["separators"],
              length_function=len,
          )
  
      def get_retriever(self):  # 获取向量存储对象
          return self.vector_store.as_retriever(search_kwargs={"k": chroma_conf["k"]})
  
      def load_document(self):  # 加载文档函数，调用此函数会自动加载需求文档
          def check_md5_hex(md5_for_check):  # 检测是否已加载过该文档
              if not os.path.exists(get_abs_path(chroma_conf["md5_hex_store"])):
                  open(get_abs_path(chroma_conf["md5_hex_store"]), "w", encoding="utf-8").close()
                  return False
              with open(get_abs_path(chroma_conf["md5_hex_store"]), "r", encoding="utf-8") as f:
                  for line in f.readlines():
                      line = line.strip()
                      if line == md5_for_check:
                          return True
              return False
  
          def save_md5_hex(md5_for_save):  # 标记已加载过的文档
              with open(get_abs_path(chroma_conf["md5_hex_store"]), "a", encoding="utf-8") as f:
                  f.write(md5_for_save+"\n")
  
          def get_file_documents(read_path: str):  # 加载文档
              if read_path.endswith("txt"):
                  return txt_loader(read_path)
              elif read_path.endswith("pdf"):
                  return pdf_loader(read_path)
              elif read_path.endswith("csv"):
                  return csv_loader(read_path)
              else:
                  return []
  
          allowed_files_path = listdir_with_allowed_type(  # 获取指定后缀的文件
              get_abs_path(chroma_conf["data_path"]),
              tuple(chroma_conf["allow_knowledge_file_type"])
          )
  
          for path in allowed_files_path:
              md5_hex = get_file_md5_hex(path)
              if not md5_hex:  # 处理MD5计算失败的情况
                  logger.warning(f"[加载知识库] {path} MD5计算失败，跳过")
                  continue
  
              if check_md5_hex(md5_hex):
                  logger.info(f"[加载知识库] {path} 内容已经存在于知识库，跳过")
                  continue
              try:
                  documents: list[Document] = get_file_documents(path)
                  if not documents:
                      logger.warning(f"[加载知识库] {path} 无有效文本内容，跳过")
                      continue
                  split_document: list[Document] = self.spliter.split_documents(documents)
                  if not split_document:
                      logger.warning(f"[加载知识库] {path} 分片后无内容，跳过")
                      continue
  
                  self.vector_store.add_documents(split_document)
                  save_md5_hex(md5_hex)
                  logger.info(f"[加载知识库] {path} 内容加载成功")
              except Exception as e:
                  logger.exception(f"[加载知识库] {path} 加载失败")
                  continue
  
  
  # for testing
  if __name__ == '__main__':
      store = VectorStoreService()
      store.load_document()
      retriever = store.get_retriever()
      res = retriever.invoke("迷路")  # 进行相似度搜索
      for r in res:
          print(r.page_content)
          print("-" * 20)
  
• 总结服务 rag_service.py 用户提问后，搜索参考资料，调用模型，将参考内容总结后进行输出