LangChain

大模型的基本介绍以及调用

大模型基本介绍

1.什么是大模型应用

• 统应用：是由程序员告诉计算机规则（编程），计算机照着规则执行。

  • 擅长：规则清楚、流程固定的事情；可以确保100%准确；行为可控、可追溯

  • 不擅长：没有明确规则的事情；自然语言的理解；模糊的判断和表达

• 大模型：计算机通过大量数据训练，自己学会规律和知识

  • 擅长：理解和生成自然语言；模糊问题的合理回答；总结、改写、对话、创作

  • 不擅长：准确的计算；固定的流程和规则；稳定可预测的结果

而大模型应用则是把两者的能力结合：大模型负责“思考”，传统程序负责“行动”。

例如，点外卖的功能，我们可以这样划分：

• 菜价、优惠、支付 → 传统程序

• “给我推荐点清淡的” → 大模型

• 最终下单、扣钱 → 传统程序

综上所述，大模型应用就是整合传统程序和大模型的能力和优势来开发的一种应用。

另外，我们熟知的AI对话产品，比如通义千问、豆包这样的APP或者聊天机器人，也都属于大模型应用：

• 收集网页用户输入文本、上传的文件、图片 → 传统程序

• 分析和理解用户输入的问题 → 大模型

• 联网搜索与问题相关的资料 → 传统程序

• 根据资料生成答案 → 大模型

模型本身只具备理解、推理、生成回复的能力。我们平常使用的AI对话产品除了生成和推理，还有会话记忆功能、联网功能等等。这些都是大模型不具备的。是需要通过额外的程序来实现的，也就是基于大模型开发应用。

所以，我们现在接触的AI对话产品其实都是基于大模型开发的应用，并不是大模型本身，这一点大家千万要区分清楚。

2.常见的大模型

大模型	对话产品	公司	地址
GPT-3.5、GPT-4o	ChatGPT	OpenAI	https://chatgpt.com/
Claude 3.5	Claude AI	Anthropic	https://claude.ai/chats
DeepSeek-R1	DeepSeek	深度求索	https://www.deepseek.com/
文心大模型3.5	文心一言	百度	https://yiyan.baidu.com/
星火3.5	讯飞星火	科大讯飞	https://xinghuo.xfyun.cn/desk
Qwen-Max	通义千问	阿里巴巴	https://tongyi.aliyun.com/qianwen/
Moonshoot	Kimi	月之暗面	https://kimi.moonshot.cn/
Yi-Large	零一万物	零一万物	https://platform.lingyiwanwu.com/

3.大模型服务

业开发大模型应用，首先需要有一个可访问的大模型，通常有两种选择：

• 使用开放大模型

• 部署私有大模型

使用开放大模型API的优缺点如下：

• 优点：

  • 没有部署和维护成本，按调用收费

• 缺点：

  • 依赖平台方，稳定性差

  • 长期使用成本较高

  • 数据存储在第三方，有隐私和安全问题

部署私有模型：

• 优点：

  • 数据完全自主掌控，安全性高

  • 不依赖外部环境

  • 虽然短期投入大，但长期来看成本会更低

• 缺点：

  • 初期部署成本高

  • 维护困难

接下来，我们给大家演示下两种部署方式：

• 公共大模型

• 私有大模型（在本机演示，将来在服务器也是类似的）

通常发布大模型的官方、大多数的云平台都会提供开放的、公共的大模型服务。大模型官方前面讲过，我们不再赘述，这里我们看一些国内提供大模型服务的云平台：

云平台	公司	地址
DeepSeek	DeepSeek	https://www.deepseek.com
阿里百炼	阿里巴巴	https://bailian.console.aliyun.com
腾讯TI平台	腾讯	https://cloud.tencent.com/product/ti
千帆平台	百度	https://console.bce.baidu.com/qianfan/overview
SiliconCloud	硅基流动	https://siliconflow.cn/zh-cn/siliconcloud
火山方舟-火山引擎	字节跳动	https://www.volcengine.com/product/ark

这些开放平台并不是免费，而是按照调用时消耗的token来付费，每百万token通常在几毛~几元钱，而且平台通常都会赠送新用户百万token的免费使用权。（token可以简单理解成你与大模型交互时发送和响应的文字，通常一个汉字2个token左右）

开发环境准备

python的环境管理方案有很多种，例如：

• pip

• uv

• conda

这里安装UV

英文文档：https://docs.astral.sh/uv/

中文文档：https://uv.doczh.com/

UV亮点

1.直接pip即可安装

简易安装

pip install uv

全局安装

irm https://astral.sh/uv/install.ps1 | iex

2.添加国内镜像源,打开CMD输入即可

setx UV_DEFAULT_INDEX "https://pypi.tuna.tsinghua.edu.cn/simple"  

常见的国内镜像站点有：

阿里云

https://mirrors.aliyun.com/pypi/simple/

腾讯云

https://mirrors.cloud.tencent.com/pypi/simple/

火山引擎

https://mirrors.volces.com/pypi/simple/

华为云

https://mirrors.huaweicloud.com/repository/pypi/simple/

清华大学

https://pypi.tuna.tsinghua.edu.cn/simple/

中国科学技术大学

https://pypi.mirrors.ustc.edu.cn/simple/

3.使用PyCharm创建项目

也可以使用uv init创建项目，在CMD打开输入即可

使用python调用OpenAI

可以使用pip和uv安装

• 使用pip安装：

  • pip install openai
    

• 使用uv安装：

  • uv add openai
    

接下来，就可以使用SDK调用任何兼容OpenAI规范的模型了，只要将base_url和api_key设定为对应的模型提供者的url和api_key即可：

from openai import OpenAI
client = OpenAI(
    api_key="sfxxxxx",
    base_url="https://api.deepseek.com"
)

print(" 正在调用大模型...")
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一名友好的AI助教。"},
        {"role": "user", "content": "你好，你是谁?"}
    ],
    stream=False
)

print(response)

注意：将api_key直接写在代码中非常危险，所以通常我们都将其写入环境变量，程序运行时加载。

第一步 配置环境变量

在项目根目录创建一个 flie .env文件：

.env配置的API_KEY

# deepseek
DEEPSEEK_API_KEY=sk-1234567890

第二步，安装python-dotenv。

在项目中，我们通过python-dotenv库来读取环境变量，所以要先安装依赖。

uv add python-dotenv

安装成功后，会在pyproject.toml中看到新添加的依赖：

[project]
name = "lc-course"
version = "0.1.0"
description = "Add your description here"
requires-python = ">=3.13"
dependencies = [
    "notebook>=7.5.5",
    "openai>=2.28.0",
    "python-dotenv>=1.2.2",
]

第三步，读取环境变量。

from openai import OpenAI
from dotenv import load_dotenv
import os


# 加载环境变量
load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

print(" 正在调用大模型...")
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一名友好的AI助教。"},
        {"role": "user", "content": "你好，你是谁?"}
    ],
    stream=False
)

print(response)

LangChain基本介绍

认识LangChain

LangChain并不仅仅是一个框架，而是一整个智能体开发平台，包含很多不同的组件。

其中，包含一系列开源的**智能体（Agent）**开发框架，而且兼容Python和TypeScript两种语言：

LangChain是智能体开发平台，包含一套各种帮助开发、测试、评估智能体的框架。核心包括：

• LangChain：用于快速构建智能体，可兼容任何模型提供商。

• LangGraph：从底层一步步控制智能体的构建，包括记忆（Memory）、人机协同（HITL）等

• Deep Agents：用于构建复杂的、处理多步骤的任务的智能体

• LangSmith：用于测试、观察、评估、部署智能体

什么是Agent

什么是Agent，这其实没有一个标准的答案，每个人都有自己的理解。

对于这个问题，LangChain创始人Harrison Chase一个偏向技术性的答案：

An AI agent is a system that uses an LLM to decide the control flow of an application.

Agent是一种使用大语言模型（LLM）来决定应用程序控制流的系统。

在人工智能领域，Agent（通常翻译为智能体或代理）是指一种能够感知环境、进行推理、自主决策并采取行动以实现特定目标的智能系统。

特性	传统聊天机器人/LLM	AI Agent
交互模式	被动响应，问一句答一句	主动规划，以目标为导向
执行力	停留在文本生成层面	能操作软件、发送邮件、分析数据
自主性	需要人类给出详细步骤	只需给定最终目标，自主寻找路径

如果说大模型（LLM）是“大脑”，那么 Agent 就是**“拥有手脚和思维逻辑的独立个体”**。它不再只是被动地回答问题，而是能主动拆解任务并调用各种工具来完成工作。

例如：要开发一个《AI旅游助手》的应用。

如果是传统LLM应用，程序流程是这样的：

1. 用户提出需求，例如：帮我计划一个5天的北京之旅，预算8000元，我喜欢历史。
2. 调用LLM，分析用户需求，直接由LLM生成一个简单旅游计划

这个计划基于它训练数据中的通用知识，可能没有考虑当前的天气、景点是否关闭、门票是否可预订等实时信息。

如果是Agent应用，Agent可以自主规划程序流程：

1. 用户提出需求，例如：帮我计划一个5天的北京之旅，预算8000元，我喜欢历史。

2. Agent分析用户需求，分步执行：

   1. 规划： 将大目标分解为：查询机票酒店价格 -> 查询天气和景点信息 -> 设计每日行程 -> 计算总预算。

   2. 调用工具：

      1. 调用机票/酒店API，查询用户指定日期范围内的价格和可选酒店。
      2. 调用天气预报API，查询未来5天北京的天气，建议携带的衣物。
      3. 调用搜索引擎/景点API，查询故宫、国博等热门景点的最新开放时间、预约政策和当前展览。

   3. 感知与反馈： 综合感知所有查询到的实时信息，生成一个动态的、可执行的计划。例如：“根据预算和您对历史的兴趣，我推荐入住胡同里的XX酒店。第一天去故宫，但请注意下周一故宫闭馆，所以调整到第二天……总花费预计7500元，还在预算内。需要我现在帮您预订酒店和机票吗？”

Agent通过主动规划任务流程，主动使用工具，整合了实时信息，并进行了动态调整，最终产出的是一个真正可落地的方案。

总结如下：

• LLM = 聪明的大脑

• Agent = 聪明的大脑 + 手脚

当然，Agent的模式也是在不断演进的：

• 阶段一：ReAct + Tool Calling

• 阶段二：Reflection + Long Memory

• 阶段三：Multi Agent System，MAS

LangChain火速入门

准备工作

1.安装LangChain必须先安装依赖

uv add langchain

LangChain支持各种不同的模型，而且提供了对应的兼容SDK，不过也都需要安装对应依赖，你可以按需添加：

# 集成 DeepSeek
uv add langchain-deepseek

# 集成 OpenAI
uv add langchain-openai

# 集成 Anthropic
uv add langchain-anthropic

开发Agent了，基本步骤如下：

1. 加载环境变量
2. 定义工具
3. 定义Agent
4. 调用Agent

Langchain提供了create_agent方法用来快速创建Agent，我们只需要提供好Agent所需的模型（Models）、**工具（Tools）**即可。

示例代码如下

# 1.加载环境变量
from dotenv import load_dotenv

load_dotenv()

# 2.定义工具，基础版，通过注释描述工具
@tool
def getWeather(location: str) -> str:
    """
    Get the weather in a given location.
    Args:
        location: city name or coordinates
    """
    return f"Current weather in {location} is sunny"


# 3.定义Agent
agent = create_agent(
    "deepseek-chat", # 模型名称（必须是LangChain支持的模型）
    tools=[getWeather] # 工具集
)

# 4.调用模型
print("正在调用大模型...")
response = agent.invoke({
    "messages": [
        {"role": "user", "content": "杭州今天天气如何?"}
    ]
})

# 5.打印结果
print(response)

大模型不具备查询天气的能力，所以无法回答天气问题。但是，当我们提供了一个查询添加的Tool以后，它就能自动查询天气来回答问题，是不是很神奇。

那么，Agent是如何做到的呢？

Agent原理

传统的LLM应用都是一问一答的形式，模型只能根据自己的训练数据来回答，流程非常简单：

而智能体则可以调用工具与外界交互，获取实时信息，工作流程则要复杂很多，是这样的：

流程如下：

1. 用户提问（Input）：杭州今天天气如何？

2. 模型分析（Reasoning）：用户询问杭州天气，我不知道，需要调用查询天气的工具get_weather

3. 调用工具（Action）：调用工具，get_weather，传入城市"杭州"

4. 分析结果（Observation）：工具返回结果，模型分析结果，判断是否足以回答用户问题

   1. 是：整理生成响应结果
   2. 否：重复前面步骤

5. 生成结果（Output）：根据工具的结果生成响应给用户

在大模型提供的API接口中，有一个tools参数，描述了工具的详细信息：

所以，LangChain会帮助我们把tool的信息封装为此tool参数，与message一起发送给大模型，大模型就了解tool的详细信息，根据用户需求判断是否需要调用tool，需要调用哪个tool.

**模型确实不能直接调用tool，只能返回字符串。但是它可以把要调用的tool信息、参数信息都以Json格式返回：

这样一来，LangChain就会帮我们解析响应结果中的Function信息，也就是tool信息，就知道了要调用哪个函数，以及参数是什么了。LangChain就会执行该函数，再把得到的结果再次发送给大模型。

大模型的工作流程

• Model：负责推理分析、思考，相当于Agent的大脑

• Tools：负责执行任务，相当于Agent与外界交互的手脚

Agent Models(模型)

**文档:**https://docs.langchain.com/oss/python/integrations/chat

完整叫法是大语言模型（LLM）。它能够理解人类语言，使用人类语言生成内容、翻译、提取摘要、回答问题等。

不仅如此，现在大多数的模型还有一些特别能力：

• Tool calling - 调用外部工具（例如查询数据库或调用 API），并在其回复中使用这些工具返回的结果。

• Structured output - 将模型的响应结果约束为遵循已定义的格式，例如：json

• Multimodality - 可以处理和返回文本以外的数据，如图像、音频和视频。

• Reasoning - 模型可以执行多步推理来得出结论。

可以说LLM就是Agent的大脑，是Agent的推理引擎。它驱动Agent做出每个决定：何时调用工具、调用哪个工具、如何解释结果，以及何时提供最终答案。

LangChain支持现在市面上大部分的大语言模型（LLM），并且提供了统一的模型调用接口。使您可以轻松访问许多不同的模型提供者，并且在模型之间进行试验和切换也变得很容易。

langchain提供了两种常见方法用来初始化模型：

• 使用init_chat_model函数，由langchain自动创建模型对象

• 使用不同模型对应的Model类，手动创建模型对象

init_chat_model

在LangChain中开始使用独立模型的最简单方法是使用init_chat_model函数。

调用init_chat_model函数时，你需要从langchain支持的模型提供者（Model Provider）中选择一个模型，而langchain会自动初始化这个模型，非常方便。

例如，我们要使用Deepseek这个模型。

• 首先，我们需要安装模型依赖：

  • uv add langchain-deepseek
    

• 然后，我们要确保在项目的**.env环境中配置好api_key**:

• 最后，就可以直接使用init_chat_model初始化模型了：

# 导入Langchain的初始化模型的函数
from langchain.chat_models import init_chat_model
# 加载环境变量
from dotenv import load_dotenv
load_dotenv()

# 调用init_chat_model函数初始化模型，参数model用来指定模型名称，Langchain会根据模型名字自动设定base_url，并从环境变量中获取api_key
model = init_chat_model(model="deepseek-chat")


print(type(model)) # <class 'langchain_deepseek.chat_models.ChatDeepSeek'>

如果要切换其它模型，我们只需要安装其它模型依赖，然后配置API_KEY，改变模型名称即可，其它代码不用动。

init_chat_model接入DeepSeek

配置

DEEPSEEK_API_KEY=sk-.....
DEEPSEEK_BASE_URL=https://api.deepseek.com

代码

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)

print(type(model))
# 测试调用
res = model.invoke("你好")
print(res.content)

自定义模型及参数

init_chat_model默认会根据模型名称自动确定模型的提供者、其base_url，并从env读取api_key，但前提是必须是langchain支持的模型提供者(支持模型参考链接)，例如：

• Openai

• Deepseek

• Google

• Anthropic

• ...

对于其它不支持的模型，我们必须自定义模型参数来访问。

例如，我们要访问阿里云百炼的qwen-max，它就是不被langchain支持的模型，我们必须自定义模型参数来访问。

• 我们需要在环境变量中定义api_key和base_url

• 然后在init_chat_model中指定model、model_provider、base_url和api_key

• **首先，**在.env中配置好api_key和base_url：

DASHSCOPE_API_KEY=sk-915a82ea621f412ed9
DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

代码

# 非支持模型无法自动加载环境遍历，我们需要自己加载环境变量中的base_url和api_key
import os

from langchain.chat_models import init_chat_model

base_url = os.getenv("DASHSCOPE_BASE_URL")
api_key = os.getenv("DASHSCOPE_API_KEY")


# 初始化模型
model = init_chat_model(
    model="qwen-max", # 模型名称，这里可以自定义，我们用的是阿里的qwen-max
    model_provider="openai", # 如果是Langchain不支持的模型，需要指定模型提供者（虽然我们用的是阿里，但是阿里兼容openai，所以这里用openai，就是默认采用openai的API规范）
    base_url=base_url,
    api_key=api_key
)

print(type(model)) # <class 'langchain_openai.chat_models.base.ChatOpenAI'>

可见，通过参数自定义模型时，模型的类型由model_provider参数类决定。

除了修改模型提供者以外，init_chat_model方法允许我们调整模型参数，例如：

• temperature: 控制生成文本的随机性，值越小越确定，值越大越随机

• max_tokens: 控制生成文本的最大长度

• top_p: 控制生成文本的多样性，值越小越多样，值越大越确定

• timeout: 控制生成文本的超时时间

• max_retries: 控制生成文本的最大重试次数

• ...

使用Model类

其实init_chat_model方法底层就是帮我们利用Model类创建对象。但只支持有限的模型。而在langchain的社区，除了langchain官方提供的Model，还有些类是社区提供，更丰富多样。

具体支持的模型，可以查看官网地址：https://docs.langchain.com/oss/python/integrations/chat

例如，我们使用社区版本的Model类来访问阿里云百炼的通义千问模型：

1. 首先，我们需要安装依赖

   1. LangChain社区依赖：

      1.  uv add langchain-community
         

   2. 阿里云百炼依赖：

      1.  uv add dashscope
         

2. 然后，我们就可以使用Model类初始化模型了

   1. from langchain_community.chat_models.tongyi import ChatTongyi
      
      # 使用Model类初始化模型
      model = ChatTongyi(
          model="qwen-plus"
          # 其它模型参数...
      )
      

3. 测试，查看生成的模型类型：

   1. print(type(model)) # <class 'langchain_community.chat_models.tongyi.ChatTongyi'>
      

调用模型的方式

LangChain提供了两个不同的方法来访问模型：

• invoke：阻塞式访问

• stream：流式访问

  import os
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  
  # 加载.env环境变量
  load_dotenv()
  
  base_url = os.getenv("DEEPSEEK_BASE_URL")
  api_key = os.getenv("DEEPSEEK_API_KEY")
  
  # deepseek-v4-pro 正确初始化
  model = init_chat_model(
      model="deepseek-v4-pro",
      model_provider="deepseek", # 原生deepseek驱动
      base_url=base_url,
      api_key=api_key
  )
  
  print(type(model))
  # 测试调用
  res = model.invoke("你好")
  print(res.content)
  

stream

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)

stream = model.stream("月亮的首都是哪里？")

# stream调用返回的结果是一个generator，方便我们循环获取结果
print(type(stream))

# 遍历stream结果，实时打印AI的回复
for chunk in stream:
    print(chunk.content, end="", flush=True)

在Agent中使用模型

1. 阻塞式调用，使用invoke方法：

import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)

from langchain.agents import create_agent

# 1.指定Model名称，由LangChain自动初始化模型
agent = create_agent(model="deepseek-v4-pro")



# 2.调用模型，需要传入一个消息列表
response = agent.invoke({
    "messages": [{"role": "user", "content": "月亮的首都是哪里？"}]
})

print(response)

stream调用

import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)

from langchain.agents import create_agent

# 1.指定Model名称，由LangChain自动初始化模型
agent = create_agent(model="deepseek-v4-pro")



for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "月亮的首都是哪里？"}]},
    stream_mode="messages"
):
    if token.content:  # Check if there's actual content
        print(token.content, end="", flush=True)  # Print token

gent的stream模式同样返回一个generator，但是其结构由stream_mode参数决定：

• messages: 返回LLM生成的每一个片段，是一个包含token和metadata的元组（Tuple）

• updates: 返回Agent运行过程中的每一次事件，例如与LLM的对话、工具的调用等

• custom: 返回通过stream writer记录的每一次自定义的输出

如果是为了流式输出AI返回的结果，使用messages模式即可。

Agent消息（Messages）

消息类型

在LangChain中，我们并不需要自己创建BaseMessage对象，LangChain已经把常见消息根据角色（Role）创建了对应的BaseMessage的子类：

• SystemMessage：role是system，代表系统消息，用于设定模型角色和交互背景

• HumanMessage：role是user，代表用户输入的消息

• AIMessage：role是assistant，代表LLM生成的响应，包含：文本、工具调用、元数据

• ToolMessage：role是tool，代表工具调用时产生的结果

import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)

#导入消息类型
from langchain.messages import HumanMessage, AIMessage
from langchain.agents import create_agent

# 创建Agent
agent = create_agent(model="deepseek-v4-pro")

# 调用Agent，发送消息
response = agent.invoke({
    "messages": [
        HumanMessage(content="你好，我是虎哥"),
        AIMessage(content="你好，虎哥，很高兴认识你。"),
        HumanMessage(content="我的名字是什么？")
    ]
})

# print(response)
for message in response['messages']:
    message.pretty_print()

多模态消息

LangChain 也支持向模型发送多模态消息，比如图片、音频、视频、文本等。但前提是必须是多模态模型才支持。

一些支持多模态的模型有：

• qwen3.5-plus

• gpt-5-nano

远程图片测试

格式

   "role": "user",
    "content": [
        {"type": "image", "url": "https://xxx.com/a.jpeg"},
        {"type": "text", "text": "这些图描绘了什么内容？"}
    ]
}

代码测试

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv

from langchain_core.messages import HumanMessage

# 加载 .env 文件中的环境变量
load_dotenv()

# 1.初始化模型
model = init_chat_model(
    model="qwen3.5-plus",  # 这里选择qwen3.5-plus，这是一个多模态模型，支持图片、文本、音频、视频
    model_provider="openai",
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    api_key=os.getenv("DASHSCOPE_API_KEY")
)

# 2.创建智能体
agent = create_agent(model=model)

# 3.组织多模态消息
multimodal_message = HumanMessage(
    content=[
        {"type": "image",
         "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"},
        {"type": "text", "text": "这些图描绘了什么内容？"}
    ])

# 4.调用Agent，发送多模态消息
for token, metadata in agent.stream({
    "messages": [multimodal_message]
}, stream_mode="messages"):
    if token.content:
        print(token.content, end="", flush=True)

本地图片测试

格式

{
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this image."},
        {
            "type": "image",
            "base64": "AAAAIGZ0eXBtcDQyAAAAAGlzb21tcDQyAAACAGlzb2...",
            "mime_type": "image/jpeg",
        },
    ]
}

所谓本地图片，就是用户上传的图片数据或者本地存在的图片，而不是图片的url地址。我们需要将图片数据转换成base64字符串，然后发送给模型。

代码

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
import os
import base64
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage

# 加载 .env 文件中的环境变量
load_dotenv()

# 1.初始化模型
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")
)

# 2.创建智能体
agent = create_agent(model=model)

# 3.读取本地图片文件，转为 base64 编码
with open("E:\\图片\\报纸墙 粉色大波浪少女 白猫 4k动漫壁纸_彼岸图网.jpg", "rb") as f:  # 替换为你的图片路径
    img_bytes = f.read()
img_b64 = base64.b64encode(img_bytes).decode("utf-8")

# 4.组织多模态消息
multimodal_question = HumanMessage(content=[
    {
        "type": "image_url",
        "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"},
    },
    {"type": "text", "text": "帮我看看这张图片"}

])

# 5.调用Agent，发送消息
response = agent.invoke(
    {"messages": [multimodal_question]}
)

print(response['messages'][-1].content)

Agent提示词（Prompts）

• 发送给大模型的所有消息都可以称为提示词（Prompt），它直接影响模型的输出结果**。**

  其中，SystemMessage尤为重要，我们把SystemMessage称为系统提示词（System Prompt），它可以给模型设定角色和本次聊天的背景，对模型生成的内容有很大的影响。

系统提示词

在创建智能体时，我们可以直接设定system prompt，不必在每次发送消息时指定。

from langchain.agents import create_agent
from langchain.messages import HumanMessage
import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model


# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)

# 创建智能体
agent = create_agent(
    model = "deepseek-chat",
    system_prompt="像海盗一样说话."
)

for token, metadata in agent.stream(
    {"messages": [HumanMessage(content="你是谁？")]},
    stream_mode="messages"
):
    print(token.content, end="", flush=True)

提示词工程

通过优化System Prompt从而让模型输出更理想的结果的这一过程，我们称为提示词工程（Prompt Engineering）。

也就是说，提示词优化不是一锤子买卖，而是一个不断优化、测试、再优化的过程。那么，提示词到底该怎么写呢？

从内容来说，提示词通常包含以下几个部分，通常按此顺序排列：

• 身份（Identity）：描述AI的职责、沟通风格和总体目标。

• 说明（Instructions）：请指导模型如何生成所需的响应。它应该遵循哪些规则？模型应该做什么，以及模型绝对不能做什么？

• 示例（Examples）：提供可能的输入示例，以及模型期望的输出。

• 背景信息（Context）：向模型提供生成响应所需的任何额外信息，例如RAG的额外知识库数据，或您认为特别相关的任何其他数据。

从格式来说，在编写System Prompt时，您可以使用Markdown格式和XML 标签的组合来帮助模型理解提示和上下文数据的逻辑边界。

• Markdown 的标题和列表有助于标记提示的不同部分，并向模型传达层级结构。它们还可以提高开发过程中提示的可读性。

• XML 标签可以帮助明确区分一段内容（例如用作参考的辅助文档、对话示例等）的起始和结束位置。

设定角色和详细指令

角色可以帮助模型认清自己的身份，以对应的身份来回答问题。

指令则告诉模型需要遵循哪些规则，应该做什么，不应该做什么

例如：

from langchain.agents import create_agent
from langchain.messages import HumanMessage
import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model


# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)


system_prompt = """
# 身份
- 你是一个编程助手，你帮助用户编写Python代码。

# 指令
- 定义变量时，使用snake case命名法，而不是camel case命名法。
- 不要返回markdown格式说明，仅仅返回代码即可。

"""

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

for token, metadata in agent.stream(
    {"messages": [HumanMessage(content="怎样定义string变量记录学校名字，例如`黑马程序员`")]},
    stream_mode="messages"
):
    print(token.content, end="", flush=True)

输出结果：

school_name = "黑马程序员"

Few-Shot examples

有的时候我们希望模型按照固定的风格来回答问题，而这种风格又不太好描述，那我们就可以通过举例的方式让模型学习例子来回答。

用户只需在输入提示（Prompt）中提供几个输入-输出示例，模型就能理解任务模式并生成符合预期的输出：

from langchain.agents import create_agent
from langchain.messages import HumanMessage
import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model


# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)


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

# 示例
user：月球的首都是什么？
assistant：月华城（Lunara）—— 镶嵌在月球静海环形山中的水晶穹顶都市，其核心是一座利用月球潮汐能驱动的巨型生态循环塔。

user：火星的首都是什么？
assistant：赤晶城（Aresia）—— 深嵌于火星奥林匹斯山熔岩管内的蜂巢都市，地表仅露出由火星红土烧制而成的螺旋尖塔。
"""

# 创建智能体
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)

结果：

熔金城（Aurum）——悬浮于硫酸云层之上的宏伟浮空都市，以反光性合金铸造，永恒折射着昏黄的日光。

结构化输出

由于传统程序识别结构化的数据会更加方便，所以有时候我们希望LLM也能输出固定结构的内容，方便我们解析。这同样可以通过系统提示词来实现。

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

# 指令
- 请务必以JSON格式输出，不要加任何markdown样式。

# 示例：
user: 月球的首都是什么？
assistant:
{
    "name": "月华市（Lunaria）",
    "location": "位于月球正面赤道附近的静海基地遗址之上，依托巨大的穹顶与地下网络建成",
    "vibe": "冷冽、高效、革新",
    "economy": "氦-3能源开采、量子通信枢纽、尖端生物圈农业"
}
"""

agent = create_agent(
    model="deepseek-chat",
    system_prompt=system_prompt
)

response = agent.invoke(
    {"messages": [HumanMessage(content="金星的首都是什么?")]},
)

print(response['messages'][-1].content)

输出结果：

{
    "name": "硫磺城（Sulfura）",
    "location": "悬浮于金星浓厚大气层中距地表约50公里的高空，由巨大的反重力浮空平台群构成",
    "vibe": "高压、炽热、坚韧",
    "economy": "大气资源提炼（二氧化碳、硫酸）、极端环境材料制造、太阳能巨型阵列"
}

在LangChain中，实现结构化输出会更加简单。我们无需自己在提示词中添加描述实现结构化输出，而仅仅是设定好一个数据类型即可

首先，我们定义一个类，用来封装模型要输出的数据：

from pydantic import BaseModel
class CapitalInfo(BaseModel):
    name: str
    location: str
    vibe: str
    economy: str

然后，我们就可以在创建Agent时设定好输出格式：

from pydantic import BaseModel
class CapitalInfo(BaseModel):
    name: str
    location: str
    vibe: str
    economy: str


from langchain.agents import create_agent
from langchain.messages import HumanMessage
import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model


# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)


# 然后，我们就可以创建智能体并设置结构化输出的格式了。
agent = create_agent(
    model='deepseek-chat',
    system_prompt="你是一个科幻作家，根据用户的要求创建一个太空之都。",
    response_format=CapitalInfo # 设置结构化输出的格式
)

response = agent.invoke(
    {"messages": [HumanMessage(content="月球的首都是什么?")]}
)

print(response['messages'][-1].content)

Agent 工具(Toos)

工具的基本使用

一个完整的Agent至少要包含两个关键的部分：

• 模型：是Agent的大脑，负责推理、分析，规划任务步骤

• 工具：是Agent的手脚，负责执行任务，与外界交互

我们先通过一个案例快速回顾Agent定义的步骤，以及Agent的工作原理。

定义一个带有工具的Agent分为两步：

• 定义工具

• 定义Agent，绑定工具

工具的基本使用 这里就定义了一个工具

# 1.使用tool装饰器定义工具
from langchain.tools import tool

@tool
def get_weather(location: str) -> str:
    """
    获取指定地点的天气信息。
    参数：
        location: 城市名称或经纬度坐标
    """
    return f"{location} 当前天气：晴天"



from langchain.agents import create_agent
from langchain.messages import HumanMessage
import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model


# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")


model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)


# 2.创建智能体，并绑定工具
agent = create_agent(
    model="deepseek-chat",
    tools=[get_weather]
)
# 3.调用Agent
response = agent.invoke(
    {"messages": [HumanMessage(content="杭州今天天气如何?")]},
)

for message in response['messages']:
    message.pretty_print()

流程图

由此可见，所谓的工具，本质就是一个可调用的函数，要想让Agent知道有哪些工具可调用，该如何调用这些工具，就必须把这个函数的详细信息发送给模型。包括：

• 函数名

• 函数的作用

• 函数的参数和返回值信息

所以，定义工具的时候，关键就是把这些信息描述清楚即可。

自定义工具

# 1.使用tool装饰器定义工具
from langchain.tools import tool
from langchain.tools import tool



@tool("平方根", description="计算一个数字的平方根")
def tool1(x: float) -> float:
    return x ** 0.5


from pydantic import BaseModel, Field
from typing import Literal

#celsius = 摄氏度（℃），日常国内用的温度单位
#fahrenheit = 华氏度（℉），欧美常用

# class WeatherInput(BaseModel):
# class：定义一个类
# WeatherInput：类名（你要求不改，我就没改）
# BaseModel：来自 Pydantic，作用是校验数据格式
# """查询天气的输入参数."""
# 文档字符串，说明这个类是干嘛的
# AI 会读取它
# location: str = Field(description="城市名称或坐标")
# location：参数名
# : str：类型必须是字符串
# Field(...)：给 AI 看的说明
# 意思：必须传城市名或坐标
# units: Literal["celsius", "fahrenheit"]
# Literal：表示只能二选一
# 只能传 "celsius" 或 "fahrenheit"
# 不能传别的！
# default="celsius"
# 不传这个参数时，默认用摄氏度
# include_forecast: bool
# bool：布尔类型
# 只能是 True 或 False
# default=False：默认不包含 5 天预报
class WeatherInput(BaseModel):
    """查询天气的输入参数."""
    location: str = Field(description="城市名称或坐标")   #Field 给AI看说明
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="温度单位偏好"
    )
    include_forecast: bool = Field(
        default=False,
        description="是否包含5天预报"
    )

# @tool(args_schema=WeatherInput)
# @tool：把这个函数变成 AI 能用的工具
# args_schema=WeatherInput：
# → 参数规则按照上面那个类来
# → AI 会按类的要求传参
# def get_weather(...) -> str:
# 定义函数，名字叫 get_weather
# -> str：返回字符串
# temp = 22 if units == "celsius" else 72
# 如果是摄氏度 → 22 度
# 如果是华氏度 → 72 度
# （这是模拟数据）
# f"{location} 当前天气：{temp} 度..."
# 拼接返回中文天气信息
# if include_forecast:
# 如果为 True → 追加未来 5 天预报
# return result
# 返回最终天气结果给 AI

# 工具函数：只改中文描述，其余完全不变
@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
    """获取当前天气及可选预报信息。"""
    temp = 22 if units == "celsius" else 72
    result = f"{location} 当前天气：{temp} 度 {units[0].upper()}"
    if include_forecast:
        result += "\n未来5天天气预报：晴天"
    return result


from langchain.agents import create_agent
import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model


# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)

# 2.创建智能体，并绑定工具
agent = create_agent(
    model="deepseek-chat",
    tools=[get_weather]
)
# 3.调用Agent
sqrt_result =tool1.invoke({"x": 467})
print(sqrt_result)

# 调用查询天气工具
res=get_weather.invoke({"location": "杭州", "include_forecast": False})
print(res)

预定义工具

LangChain中提供了很多预定义好的工具，方便我们使用，可使用的预定义工具列表可参考官网：

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

例如，模型本身只能根据本身的训练数据回答问题，无法获取实时信息。但如果我们给它提供了web搜索的工具，

那么你的Agent就如同具备了实时web搜索的能力，回答会更加准确。

这里用Tavily其他工具可以看文档

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

**Tavily官方文档：**https://app.tavily.com/home

配置环境变量

把这个KEY配置到我们的.env文件中：

import os
import sys
from dotenv import load_dotenv
# 解决 Windows GBK 编码无法输出 emoji 的问题
sys.stdout.reconfigure(encoding="utf-8")
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
# 使用tavily作为web搜索工具
from langchain_tavily import TavilySearch
# 加载.env环境变量 — 必须在所有依赖环境变量的代码之前调用
load_dotenv()
# 初始化工具，并设置参数，具体参数设置参考官网
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
)
base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")
model = init_chat_model(
    model="deepseek-chat",
    model_provider="openai",  # DeepSeek 使用 OpenAI 兼容 API
    base_url=base_url,
    api_key=api_key
)
agent = create_agent(
    model=model,  # 传 model 对象，不是字符串
    tools=[tool],
    system_prompt="你是一个智能助手，你使用工具来解决用户问题。"
)
# 调用工具
for chunk in agent.stream(
    {"messages": [HumanMessage(content="北京接下来5天天气如何?")]},
    stream_mode="updates"
):
    for step, data in chunk.items():
        print(f"step: {step}")
        print(f"content: {data['messages'][-1].content_blocks}")
        print()

注意，LangChain提供的TavilySearch工具描述非常复杂，参数也很多。会有额外的网络消耗。如果我们仅仅是需要query参数，建议自定义工具。

import os
import sys
from dotenv import load_dotenv

# 解决 Windows GBK 编码无法输出 emoji 的问题
sys.stdout.reconfigure(encoding="utf-8")
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
# 使用tavily作为web搜索工具
from langchain_tavily import TavilySearch
from langchain.tools import tool

from pydantic import BaseModel, Field

# 加载.env环境变量 — 必须在所有依赖环境变量的代码之前调用
load_dotenv()

# 使用tavily作为web搜索工具
tavily = TavilySearch(
    max_results=5,
    topic="general"
)

#这里就是然他指定在网上查询信息这个功能
@tool
def web_search(query: str):
    """使用百度在网上查找信息"""
    return tavily.invoke(query)



# 智能体回答所引用的网页信息
class Reference(BaseModel):
    title: str = Field(description="回答中引用的网页标题")
    url: str = Field(description="回答中引用的网页链接")

# 智能体的回答内容
class AnswerInfo (BaseModel):
    answer: str = Field(description="给到用户的最终回答")
    reference: list[Reference] = Field(description="回答中引用的网页列表")


base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)
# 创建智能体，使用预定义工具tavily
agent = create_agent(
    model="deepseek-chat",
    tools=[web_search],
    system_prompt="你是一个智能助手，你使用工具来解决用户问题。",
    response_format=AnswerInfo
)


# 调用agent
response = agent.invoke(
    {"messages": [HumanMessage(content="蒸蚌是什么梗？")]},
)

# 获取结构化输出
print(response['structured_response'])

Agent 会话记忆

会话记忆的基本介绍

模型本身是没有记忆的，它记不住历史的会话内容，

我们需要通过技术手段，帮助模型记住会话历史，产生记忆。

对于Agent而言，记忆至关重要，因为它能让代理记住之前的交互情况，从反馈中学习，并适应用户的偏好。随着代理处理的任务愈发复杂，涉及的用户交互也越来越多，这种能力对于提高效率和用户满意度而言变得不可或缺。

记忆的分类

对于智能体而言，记忆分为了两类：

• 短期记忆（short-term memory）

• 长期记忆（long-term memory)

注意，大家不要被字面上的意思误导了，很多人看到名字就误以为：短期记忆就是临时记忆，断电就没了；长期记忆就是永久记忆，持久保存。

对于智能体而言，这是完全错误的理解！！！

简单用一句话概括的话：

• 短期记忆：当前任务或会话的上下文（Working Memory 或 Session Memory）

• 长期记忆：跨任务或会话的经验与知识（Persistent Memory）

比如，一个公司数据分析的Agent。

用户提出需求：

“帮我写Q1的销售分析报告”

Agent：

短期记忆：

• 对话历史

• 查询到Q1的销售数据

• 任务目标及执行状态

长期记忆：

• 公司的KPI算法

• 用户偏好的报告形式

总结：

	短期记忆	长期记忆
生命周期	当前会话（短暂）	跨任务、跨会话（永久）
内容	当前任务状态	知识、经验、用户偏好
是否跨任务	❌	✅
存储	Redis/内存	DB/Vector DB

短期记忆

由于短期记忆通常生命周期是当前会话，所以我们也可以称为会话记忆。Agent的会话记忆通常包含三部分：

• 对话历史

• 查询结果

• 任务状态

对于简单的Agent来说，任务没有做拆分，也就不需要记录任务状态，只用考虑会话历史和查询结果就可以了。后续我们会学习如何自定义更复杂的Agent会话记忆。

LangChain提供了自动化的记忆管理方案：

• 首先，LangChain把会话记忆（也就是Messages列表）记录为AgentState的一部分

• AgentState通过Checkpointer对象来保存，每一次与AI的交互都会生成一个快照，记录为一个checkpoint，把同一会话的所有checkpoint组合在一起，就是完整的会话历史了。

• 为了区分不同的会话记忆，不同会话需要设定各自的thread_id，相同会话则使用相同thread_id

• 向Agent发起会话时必须指定自己的thread_id以唤起对应的会话记忆

代码

from langchain.agents import create_agent
from langchain.messages import HumanMessage
import os
import sys
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
# langchain提供的checkpointer的默认实现，基于内存存储
from langgraph.checkpoint.memory import InMemorySaver

# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)
# 创建智能体
agent = create_agent(
    model = "deepseek-chat",
    checkpointer=InMemorySaver()
)
# 设定thread_id，作为会话标识
config = {"configurable": {"thread_id": "thread_1"}}
# 第一次调用，告知AI我的信息
response = agent.invoke(
    {"messages": [HumanMessage(content="你好，我叫虎哥，我最喜欢猫猫。")]},
    config # 调用时添加thread_id，区分不同会话
)

print(response["messages"][-1])

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

print(response["messages"][-1])

短期记忆-持久储存

• SqlLiteSaver ：基于sqlite存储

• PostgresSaver ：基于Postgres存储

• CosmosDBSaver ：使用Azure Cosmos DB的实现

  还有其他数据库如redis等

文档：https://docs.langchain.com/oss/python/langgraph/persistence#checkpointer-libraries

这里用sqlite来进行存储测试

# pip安装
# pip install langgraph-checkpoint-sqlite

# uv安装
uv add langgraph-checkpoint-sqlite

代码

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
import os
import sys
import sqlite3
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
# langgraph提供的sqlite持久化存储
from langgraph.checkpoint.sqlite import SqliteSaver

# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

# 初始化checkpointer（sqlite持久化存储，重启后记忆保留）
conn = sqlite3.connect("checkpoint.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
checkpointer.setup()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

# deepseek-v4-pro 正确初始化
model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek", # 原生deepseek驱动
    base_url=base_url,
    api_key=api_key
)
# 创建智能体
agent = create_agent(
    model=model,
    checkpointer=checkpointer
)
# 设定thread_id，作为会话标识
config = {"configurable": {"thread_id": "thread_2"}}
# 第一次调用，告知AI我的信息
response = agent.invoke(
    {"messages": [HumanMessage(content="你好，我叫虎哥，我最喜欢猫猫。")]},
    config # 调用时添加thread_id，区分不同会话
)

print(response["messages"][-1])

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

print(response["messages"][-1])

可以看到成功生成数据库文件

记忆管理策略

由于会话记忆要保存会话的历史，并且在调用LLM时携带历史消息列表。而当会话越来越长时，历史消息就可能超过LLM的上下文限制。例如，DeepSeek的上下文不能超过128K.

一旦会话历史超过上下文窗口，就会出现上下文丢失的情况，从而导致丢失记忆。而且即便不丢失，太长的上下文容易让模型出现“注意力分散”问题，模型的响应速度、回答质量会大大降低。

未来解决这一问题，通常有以下几种手段：

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

修剪消息

修剪消息并不是真正的删除消息，在AgentState中的消息列表依然是完整的，只不过发送给LLM之前会进行修剪，只保留一部分消息。

参考文档：https://docs.langchain.com/oss/python/langchain/short-term-memory#trim-messages

删除消息

删除消息与修剪不同：

• 修剪消息：只是从State中选取一部分消息发送给模型

• 删除消息：直接删除State中保存的消息，也就是说消息历史中不再存在！

参考文档：https://docs.langchain.com/oss/python/langchain/short-term-memory#delete-messages

总结消息

不管是修剪还是删除，都会导致一部分消息丢失，从而丢失记忆。所以就有了第三种策略：总结消息（Summarize Messages）

它的思路很简单，就是把历史的消息利用大模型总结出摘要，然后把最新的消息拼接在一起作为新的消息列表发送给大模型，这样既不会超出模型的上下文窗口限制，还能尽量保留所有的记忆。

LangChain提供了总结消息的默认实现：SummarizationMiddleware

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain_core.messages import HumanMessage
import os
import sys
import sqlite3
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
# langgraph提供的sqlite持久化存储
from langgraph.checkpoint.sqlite import SqliteSaver

# 解决 Windows 控制台 UTF-8 编码问题
sys.stdout.reconfigure(encoding='utf-8')

# 加载.env环境变量
load_dotenv()

# 初始化checkpointer（sqlite持久化存储，重启后记忆保留）
conn = sqlite3.connect("checkpoint.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
checkpointer.setup()

base_url = os.getenv("DEEPSEEK_BASE_URL")
api_key = os.getenv("DEEPSEEK_API_KEY")

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

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

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


for message in final_response["messages"]:
    message.pretty_print()

Agent入门实战（AI私厨）

功能模拟（jupyter）

1. 配置

确保你的.env中有以下配置：

# .env

# 阿里百炼
DASHSCOPE_API_KEY=sk-913a82aa121f412aa9a8c
DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

# web_search引擎
TAVILY_API_KEY=tvly-dev-Nkc1MzzM4FtWCI7ENby4

依赖

参考下面的依赖：

[project]
name = "food-recipe-recommender"
version = "0.1.0"
description = "AI-powered recipe recommender based on uploaded food images"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
    "langchain>=0.3.0",
    "langchain-community>=0.3.0",
    "langchain-openai>=0.2.0",
    "langchain-core>=0.3.0",
    "streamlit>=1.40.0",
    "pillow>=11.0.0",
    "python-dotenv>=1.0.0",
    "tavily-python>=0.5.0",
    "langchain-tavily>=0.1.0",
    "fastapi>=0.109.0",
    "uvicorn>=0.27.0",
    "python-multipart>=0.0.6",
    "alibabacloud-oss-v2>=1.2.4",
    "langgraph-checkpoint-sqlite>=3.0.3",
]

[tool.uv]
dev-dependencies = [
    "pytest>=8.0.0",
    "black>=24.0.0",
]

加载配置

首先，我们要加载环境变量：

# 加载环境变量
from dotenv import load_dotenv

load_dotenv()

定义工具

然后，我们要定义工具：

from langchain_tavily import TavilySearch

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

初始化模型

接着，初始化多模态模型：

from langchain.chat_models import init_chat_model
import os

model = init_chat_model(
    model="qwen3.5-plus",  # 模型名称，这里选择qwen3.5-plus，这是一个多模态模型
    model_provider="openai",
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    api_key=os.getenv("DASHSCOPE_API_KEY")
)

记忆管理

然后，我们定义记忆管理的checkpointer:

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# 连接sqlite
connection = sqlite3.connect("resources/personal_chief.db", check_same_thread=False)
# 初始化checkpointer
checkpointer = SqliteSaver(connection)
# 自动建表
checkpointer.setup()

初始化Agent

接下来，就是智能体了：

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
)

LangSmith联调测试（推荐）

LangChain的Agent底层是基于LangGraph实现的，而LangGraph提供了完整的后端部署功能，自带非常完善的API接口，无需我们额外处理。

同时，LangChain还提供了基于LangSmith的GUI控制台实现Agent的调试、监控、一键部署。

配置LangSmith

LangSmith提供了对Agent的GUI管理界面，而且还支持一键云部署功能。通常在测试阶段，建议大家在Agent中引入Simth，方便做测试和调试。

首先，我们要注册LangSmith，开通服务，生成API_KEY。

注册地址：https://smith.langchain.com/

注册成功后，登录，进入控制台，找到settings：

在settings页面找到API Keys菜单，创建自己的API_KEY：

接着，我们无需额外安装依赖，只需要在项目的.env文件中添加配置即可：

# langsmith
LANGSMITH_API_KEY=lsv2_pt_.......
LANGSMITH_TRACING=true
# project name
LANGSMITH_PROJECT=lc-course

开发Agent

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain.agents import create_agent
import os

# 1.加载环境变量
from dotenv import load_dotenv
load_dotenv()

# 2.web搜索工具，使用tavily作为web搜索工具
web_search = TavilySearch(
    max_results=5,
    topic="general"
)

# 3.多模态模型
model = init_chat_model(
    model="qwen3.5-plus",  # 模型名称，这里选择qwen3.5-plus，这是一个多模态模型，支持图片、文本、音频、视频
    model_provider="openai",
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    api_key=os.getenv("DASHSCOPE_API_KEY")
)

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

请严格按照流程，优先调用 web_search 工具搜索食谱，搜索不到的情况下才能自己发挥。
"""

# 5.创建Agent
agent = create_agent(
    model=model,  # 模型
    tools=[web_search],  # 工具
    system_prompt=system_prompt  # 系统提示词
)

注意：

• LangGraph会自动托管Agent的记忆，因此代码中不用自己添加checkpointer！

• LangGraph自带Restful的API接口，我们只要定义好Agent就可以，其它不用管

本地部署

我们使用LangGraph命令行在本地部署，所以要先安装LangGraph的依赖。

然后，在项目根目录添加一个langgraph配置文件：

添加下面的内容：

{
    "dependencies": ["."],
    "graphs": {
        "chief_agent": "./app/agents/personal_chief.py:agent"  
    },
    "env": ".env"
}

注意：其中的agent配置格式为：

[Agent文件路径]:[Agent变量名]

例如，在我们的配置中：

• ./app/agents/personal_chief.py：就是文件路径

• agent：就是文件中定义的Agent名字

最后，打开Pycharm终端：

使用LangGraph命令本地部署Agent：

uv run langgraph dev

当然，LangGraph也支持Docker部署方案，可参考以下链接：

https://docs.langchain.com/langsmith/deploy-with-control-plane#step-2-build-docker-image

LangSmith Studio测试

由于我们部署时配置了LangSmith，所以可以直接访问LangSmith提供的调试GUI界面：

这里可以非常方便的调试我们的Agent，查看我们Agent的运行细节：

可以直接在界面中测试：

也可以查看详细的调用过程：

同时，LangSmith还提供了一键云部署功能，可以把Agent部署到云端：

Agent测试阶段使用LangSmith吧

Agent实战开发

Agent跑通了，但目前还存在几个问题：

• 目前的图片信息还是采用base64方式提交给模型，会占用大量内存，性能差

• 我们没有开发自己的前端，用户体验不好

在向模型提交多模态消息，比如：音频、视频、图片时，我们不建议直接发送文件数据（base64）给模型，这会大量占用内存和会话记忆。更常见的方案是：

• 先将多模态文件上传至通用的OSS服务，例如：阿里云OSS、腾讯云COS等

• 获取oss服务的文件url地址，组织多模态消息，发送给大模型

因此，我们需要单独开发一个文件上传的服务接口，让前端先上传好文件，再调用Agent.

我们的服务端需要具备以下接口：

• 对话接口：接收用户聊天消息，并调用Agent

• 会话管理接口：查询或删除会话历史

• 文件上传接口：调用OSS提供的客户端，实现文件上传授权，将来由前端完成文件上传，文件不经过服务器。

FastAPI服务端

uv add fastapi alibabacloud-oss-v2

项目结构：

  app/
  ├── main.py                    # FastAPI 入口，配置路由和静态文件
  │
  ├── agents/
  │   └── personal_chief.py      # AI 代理核心逻辑
  │ 
  ├── api/
  │   └── v1/
  │       ├── chat.py             # 对话 API
  │       │   ├── POST /chat/stream     流式对话
  │       │   ├── GET  /chat/messages   获取历史
  │       │   └── DELETE /chat/messages 清空历史
  │       │
  │       └── oss.py              # OSS 上传签名 URL
  │
  ├── models/
  │   └── schemas.py              # Pydantic 数据模型，请求/响应数据结构定义  
  │
  ├── common/
  │   └── logger.py               # 日志配置
  │
  └── static/                     # Next.js 编译产出的静态网页 
      ├── index.html              # 前端入口
      ├── _next/                  # Next.js 构建资源
      └── ...                     # 其他静态资源

配置OSS配置存储

这里可以先注册阿里云： https://oss.console.aliyun.com/overview

注册登录成功后，访问链接：https://oss.console.aliyun.com/overview，即可看到oss控制台：

申请API_KEY:https://ram.console.aliyun.com/overview?activeTab=workflow

1.选择用户创建用户

2.点击：用户>创建用户，填写用户信息：

3.创建完成后，把AccessKey的ID和Secret复制到配置文件 在这里创建

4.选择权限管理新增OSS权限

5.开通OSS

访问链接：https://oss.console.aliyun.com/overview，进入oss控制台，选择**创建Bucket**：

进入具体的Bucket设置，关闭公共访问开关：

设置权限为公共读：

接着，我们要开启跨域访问权限：

然后配置文件编写

OSS_ACCESS_KEY_ID=LTBI5tHzjC36KhCJfPqlbaCo
OSS_ACCESS_KEY_SECRET=aDPGBi1nIlYzcEmk5djWJGUv3w9Qkh
OSS_BUCKET=ceshi11retyui11111    #Bucket 名称

目前，新建的Bucket还是无法访问的，是私有的。为了方便测试，这里我们暂时将其设置为公共读。

注意：

• 实际开发中oss中的图片应该设置为私有！不可对外暴露！！由额外的CDN服务对外暴露！

• 本例中，我们为了方便暂时将bucket设置为公共读，测试完毕后请及时关闭权限！

后端

agents.personal_chief.py

from langchain.agents.middleware import SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessageChunk, AIMessage
from langchain_tavily import TavilySearch
from langchain.agents import create_agent

from app.common.logger import logger
import os
import asyncio
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
# 加载环境变量
from dotenv import load_dotenv

#加载环境变量的配置文件
load_dotenv()
# web搜索工具，使用tavily作为web搜索工具
# max_results = 2,   最多返回 2 条搜索结果
# topic=general
# Tavily 支持 3 种类型：
# general：通用（默认，新闻、知识、问答都可以）
# news：只搜新闻
# qa：只搜问答类精准结果
tavily = TavilySearch(
    max_results=2,  #最多返回 2 条搜索结果
    topic="general"
)

# 多模态模型
model = init_chat_model(
    model="qwen3.5-plus",
    model_provider="openai",
    base_url=os.getenv("DASHSCOPE_BASE_URL"),  #定义的配置的URL地址
    api_key=os.getenv("DASHSCOPE_API_KEY"),  #定义的API KEY
    streaming=True  # 启用流式输出，模型逐 token 返回
)

# 初始化checkpointer
#这里用的是本地 sqlite数据库
db_dir = "db"  # 定义数据库文件夹
os.makedirs(db_dir, exist_ok=True)  #创建文件夹（不存在就创建）
#3. 连接 SQLite 数据库文件
# check_same_thread=False（非常关键）
# FastAPI 是多线程运行的
# 默认 SQLite 只允许创建它的线程使用
# 加这个参数 = 允许多线程共用数据库
# 不加：接口会直接报错！
conn = sqlite3.connect(os.path.join(db_dir, "personal_chief.db"), check_same_thread=False)
#4. 创建 SQLite 记忆存储器
checkpointer = SqliteSaver(conn)

#5. 初始化数据库表
checkpointer.setup()


# 初始化中间件：当消息数超过 10 条时自动总结历史会话，保留最近 4 条维持连续对话上下文
middleware = SummarizationMiddleware(
    model="deepseek-v4-pro",
    trigger=("messages", 10),  # 消息数超过 10 条触发总结（约 4-5 轮对话）
    keep=("messages", 4)       # 总结后保留最近 4 条消息
)

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

请严格按照流程，优先调用 web_search 工具搜索食谱，搜索不到的情况下才能自己发挥。
"""



# 创建代理
agent = create_agent(
    model=model,  # 模型
    tools=[tavily],  # 工具
    checkpointer=checkpointer,  # 记忆
    middleware=[middleware],  #对应会话进行一个总结
    system_prompt=system_prompt  # 系统提示词
)


# 流式对话
async def search_recipes(prompt: str, image: str, thread_id: str):
    """调用agent搜索食谱"""
    logger.info(f"[用户]: {prompt}, image: {image}, thread_id: {thread_id}")
    try:
        # 判断是否有图片，封装不同格式的消息
        #这里消息有几种 如下
        # - SystemMessage：role是system，代表系统消息，用于设定模型角色和交互背景
        # - HumanMessage：role是user，代表用户输入的消息
        # - AIMessage：role是assistant，代表LLM生成的响应，包含：文本、工具调用、元数据
        # - ToolMessage：role是tool，代表工具调用时产生的结果
        if not image or image.strip() == "":
            #这里如果没有就返回纯文本消息
            message = HumanMessage(content=prompt)
        else:
            #有图片 → 图文混合消息（多模态）
            message = HumanMessage(content=[
                {"type": "image_url", "image_url": {"url": image}},
                {"type": "text", "text": prompt}
            ])

        # 流式调用Agent，按块输出
        for chunk, metadata in agent.stream(  #让 AI 智能体（agent）开始干活，并且一段一段把结果流式吐出来，而不是等全部做完才返回。
            {"messages": [message]},   # 第1个参数：给AI传什么消息
            {"configurable": {"thread_id": thread_id}},  # 第2个参数：会话记忆
            stream_mode="messages"   # 第3个参数：流式返回格式
        ):
            # 检测工具调用：提示用户当前在搜索
            if isinstance(chunk, AIMessageChunk):  # 判断是不是 AI 返回的消息块
                if chunk.tool_calls:  # 如果 AI 正在调用工具（搜索）
                    yield "\n🔍 正在搜索食谱…\n"
                elif chunk.content: #如果 AI 返回了文字内容
                    yield chunk.content
            await asyncio.sleep(0)  # 让出事件循环，确保数据立即推送到客户端

    except Exception as e:
        import traceback
        logger.error(f"\n[错误]: {str(e)}\n{traceback.format_exc()}")
        yield "信息检索失败，试试看手动输入食物列表？"


# 清空会话
def clear_messages(thread_id: str):
    """清空会话"""
    logger.info(f"清空历史消息，thread_id: {thread_id}")

    # checkpointer = 之前创建的数据库记忆管理器
    # delete_thread(thread_id) = LangGraph
    # 内置方法
    # 作用：把这个
    # thread_id
    # 对应的所有对话历史，从
    # SQLite
    # 数据库里彻底删除
    checkpointer.delete_thread(thread_id)


# 查询会话历史
def get_messages(thread_id: str) -> list[dict[str, str]]:
    """获取会话历史"""
    logger.info(f"获取历史消息，thread_id: {thread_id}")

    checkpoint = checkpointer.get({"configurable": {"thread_id": thread_id}})    #从数据库读取会话状态 固定语法

    if not checkpoint:  #如果没有记录 → 返回空列表
        return []

    channel_values = checkpoint.get("channel_values")      #$ 取出 channel_values（状态数据）
    if not channel_values:
        return []

    messages = channel_values.get("messages", [])  #取出 messages（真正的聊天记录）
    if not messages:
        return []

    result = []
    for msg in messages:
        if not msg.content:  #第一步：如果没有消息，直接返回空
            continue

        # isinstance(msg, 类型)  判断这条消息到底是用户发的，还是AI发的
        # HumanMessage 代表用户输入 格式化为："role": "user"
        # AIMessage  代表  AI回复 格式化为："role": "assistant"
        # msg.content 取出消息里的真实文字内容
        if isinstance(msg, HumanMessage):
            result.append({"role": "user", "content": msg.content})
        elif isinstance(msg, AIMessage):
            result.append({"role": "assistant", "content": msg.content})

    return result

API.v1.chat.py

import json
from fastapi import APIRouter, Request
from app.models.schemas import ChatRequest, ChatResponse
from fastapi.responses import StreamingResponse
from app.agents.personal_chief import search_recipes, get_messages, clear_messages
router = APIRouter()
def _sse_encode(chunk: str) -> str:
    """将文本块编码为 SSE 格式"""
    return f"data: {json.dumps(chunk)}\n\n"

async def _sse_stream(prompt: str, image_url: str, thread_id: str):
    """包装流式输出为 SSE 事件流"""
    async for text in search_recipes(prompt, image_url, thread_id):
        yield _sse_encode(text)
    yield "data: [DONE]\n\n"

@router.post("/chat")
async def chat(request: ChatRequest, req: Request):
    """对话接口：默认 SSE 流式输出，Header 加 Accept: application/json 返回完整 JSON"""
    accept = req.headers.get("accept", "")

    # 流式模式（默认）- SSE 协议
    if "application/json" not in accept:
        return StreamingResponse(
            _sse_stream(request.message, request.image_url, request.thread_id),
            media_type="text/event-stream; charset=utf-8"
        )

    # JSON 模式（Swagger 文档用）
    chunks = []
    async for chunk in search_recipes(request.message, request.image_url, request.thread_id):
        chunks.append(chunk)
    return {"thread_id": request.thread_id, "reply": "".join(chunks)}

@router.get("/chat/messages")
async def get_chat_messages(thread_id: str):
    """获取历史消息"""
    messages = get_messages(thread_id)
    return {"messages": messages}

@router.delete("/chat/messages")
async def clear_chat_messages(thread_id: str):
    """清空历史消息"""
    clear_messages(thread_id)
    return {"success": True}

API.v1.oss.py

import alibabacloud_oss_v2 as oss
from fastapi import APIRouter
from datetime import timedelta
import os

router = APIRouter()

# 从环境变量中加载凭证信息，用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

# 加载SDK的默认配置，并设置凭证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider

# 方式一：只填写Region（推荐）
# 必须指定Region ID，SDK会根据Region自动构造HTTPS访问域名
cfg.region = 'cn-beijing'

# 使用配置好的信息创建OSS客户端
client = oss.Client(cfg)

# OSS 域名配置
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")

    # 生成预签名 PUT URL（用于前端上传）
    put_result = client.presign(oss.PutObjectRequest(
        bucket=OSS_BUCKET,
        key=filename,
        content_type=content_type,
    ), expires=timedelta(seconds=3600))

    # 生成预签名 GET URL（用于模型访问图片，24小时有效）
    get_result = client.presign(oss.GetObjectRequest(
        bucket=OSS_BUCKET,
        key=filename,
    ), expires=timedelta(hours=24))

    return {
        "uploadUrl": put_result.url.strip('"'),
        "contentType": content_type,
        "accessUrl": get_result.url.strip('"')
    }

common.logger.py

# app/common/logger.py
import logging
import sys

# 配置日志格式：时间 - 级别 - 模块 - 消息
LOG_FORMAT = "%(asctime)s - %(levelname)s - %(name)s - %(message)s"

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format=LOG_FORMAT,
        handlers=[
            logging.StreamHandler(sys.stdout), # 输出到控制台
            # logging.FileHandler("app.log")   # 如果需要存到文件可以开启
        ]
    )

# 创建一个全局的 logger 实例
logger = logging.getLogger("personal_chief")

models.schemas.py

from typing import Optional
from pydantic import BaseModel


class ChatRequest(BaseModel):
    message: str
    image_url: Optional[str] = None
    thread_id: str


class ChatResponse(BaseModel):
    thread_id: str
    reply: str

main.py

import os
from dotenv import load_dotenv
# 加载环境变量 — 必须在所有依赖 env 的 import 之前调用！
load_dotenv()

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from app.api.v1 import chat
from app.api.v1 import oss
from app.common.logger import setup_logging

# 初始化日志配置
setup_logging()

app = FastAPI(
    title="Personal Chief API",
    description="私厨 — AI 驱动的食谱助手，支持上传食材图片、智能搜索食谱、流式对话",
    version="0.1.0"
)

# CORS 跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 挂载 API 路由
app.include_router(chat.router, prefix="/api/v1", tags=["对话"])
app.include_router(oss.router, prefix="/api/v1", tags=["OSS 上传"])

# 前端页面
STATIC_DIR = os.path.join(os.path.dirname(__file__), "static")


@app.get("/", response_class=HTMLResponse)
async def root():
    """前端入口 — 动态读取，修改 HTML 无需重启服务"""
    with open(os.path.join(STATIC_DIR, "index.html"), encoding="utf-8") as f:
        return f.read()


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host="127.0.0.1", port=8081, reload=False)

前端

<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>私厨 — AI 食谱助手</title>
  <script src="https://unpkg.com/vue@3/dist/vue.global.js"></script>
  <style>
    * { margin: 0; padding: 0; box-sizing: border-box; }
    body { font-family: -apple-system, BlinkMacSystemFont, "Microsoft YaHei", sans-serif; background: #f5f0eb; }
    #app { display: flex; flex-direction: column; height: 100dvh; }
    .header { background: linear-gradient(135deg, #e07a3d, #c0392b); color: #fff; padding: 12px 16px; text-align: center; flex-shrink: 0; }
    .header h1 { font-size: 17px; }
    .header p { font-size: 11px; opacity: .8; }
    .chat { flex: 1; overflow-y: auto; padding: 14px; scroll-behavior: smooth; }
    .chat-inner { max-width: 800px; margin: 0 auto; }
    .empty { text-align: center; color: #bbb; padding-top: 60px; }
    .empty .icon { font-size: 40px; margin-bottom: 10px; }
    .msg { margin-bottom: 14px; display: flex; gap: 8px; }
    .msg.user { flex-direction: row-reverse; }
    .avt { width: 30px; height: 30px; border-radius: 50%; display: flex; align-items: center; justify-content: center; font-size: 14px; flex-shrink: 0; }
    .msg.user .avt { background: #f0c8a8; }
    .msg.assistant .avt { background: #a8e0c0; }
    .bbl { max-width: 78%; padding: 9px 13px; border-radius: 12px; font-size: 14px; line-height: 1.6; word-break: break-word; }
    .msg.user .bbl { background: #e07a3d; color: #fff; border-bottom-right-radius: 3px; white-space: pre-wrap; }
    .msg.assistant .bbl { background: #fff; border: 1px solid #e8e0d8; border-bottom-left-radius: 3px; }
    /* Markdown 渲染样式 */
    .msg.assistant .bbl h1, .msg.assistant .bbl h2, .msg.assistant .bbl h3 { margin: 10px 0 5px; font-size: 15px; color: #c0392b; }
    .msg.assistant .bbl h1 { font-size: 17px; }
    .msg.assistant .bbl h2 { font-size: 15px; }
    .msg.assistant .bbl h3 { font-size: 14px; }
    .msg.assistant .bbl p { margin: 4px 0; }
    .msg.assistant .bbl ul, .msg.assistant .bbl ol { padding-left: 18px; margin: 4px 0; }
    .msg.assistant .bbl li { margin: 2px 0; }
    .msg.assistant .bbl strong { color: #c0392b; }
    .msg.assistant .bbl em { color: #8b5e3c; }
    .msg.assistant .bbl code { background: #f5f0eb; padding: 1px 5px; border-radius: 3px; font-size: 13px; }
    .msg.assistant .bbl pre { background: #2d2d2d; color: #f8f8f2; padding: 10px 14px; border-radius: 8px; overflow-x: auto; margin: 6px 0; font-size: 13px; }
    .msg.assistant .bbl pre code { background: none; padding: 0; color: inherit; }
    .msg.assistant .bbl blockquote { border-left: 3px solid #e07a3d; padding-left: 10px; margin: 6px 0; color: #8b5e3c; }
    .msg.assistant .bbl hr { border: none; border-top: 1px solid #e8e0d8; margin: 10px 0; }
    .msg.assistant .bbl table { border-collapse: collapse; width: 100%; margin: 6px 0; font-size: 13px; }
    .msg.assistant .bbl th, .msg.assistant .bbl td { border: 1px solid #e8e0d8; padding: 4px 8px; text-align: left; }
    .msg.assistant .bbl th { background: #f8f5f0; font-weight: 600; }
    .msg.assistant .bbl a { color: #e07a3d; text-decoration: underline; }
    .msg.assistant .bbl img { max-width: 180px; border-radius: 8px; margin-top: 5px; }
    .msg.assistant .bbl .err { color: #e74c3c; }
    /* 流式输出光标 */
    .cursor { display: inline-block; width: 8px; height: 16px; background: #e07a3d; vertical-align: text-bottom; animation: blink .8s infinite; margin-left: 1px; border-radius: 1px; }
    @keyframes blink { 0%,100%{opacity:1} 50%{opacity:0} }
    /* 工具调用提示 */
    .tool-hint { display: inline-block; background: #fff3cd; color: #856404; padding: 2px 8px; border-radius: 10px; font-size: 12px; margin: 3px 0; animation: fadeIn .3s; }
    @keyframes fadeIn { from{opacity:0;transform:translateY(-3px)} to{opacity:1;transform:translateY(0)} }

    .input-bar { background: #fff; border-top: 1px solid #e8e0d8; padding: 8px 14px; flex-shrink: 0; }
    .input-inner { max-width: 800px; margin: 0 auto; }
    .thread-row { display: flex; align-items: center; gap: 6px; font-size: 11px; color: #999; margin-bottom: 6px; }
    .thread-row input { border: 1px solid #d5c8b8; border-radius: 4px; padding: 2px 5px; font-size: 11px; width: 90px; }
    .thread-row a { cursor: pointer; color: #999; text-decoration: none; }
    .thread-row a:hover { color: #e74c3c; }
    .row { display: flex; gap: 7px; align-items: flex-end; }
    .row textarea { flex: 1; border: 1px solid #d5c8b8; border-radius: 10px; padding: 7px 10px; font-size: 13px; resize: none; outline: none; font-family: inherit; height: 38px; line-height: 1.5; }
    .row textarea:focus { border-color: #e07a3d; }
    .btn { border: none; border-radius: 7px; padding: 6px 14px; font-size: 13px; cursor: pointer; font-weight: 500; white-space: nowrap; transition: .15s; }
    .btn:disabled { opacity: .5; cursor: not-allowed; }
    .btn-send { background: #e07a3d; color: #fff; min-width: 56px; }
    .btn-send:hover:not(:disabled) { background: #c0392b; }
    .btn-stop { background: #e74c3c; color: #fff; min-width: 56px; display: none; }
    .btn-stop:hover { background: #c0392b; }
    .btn-stop.show { display: inline-block; }
    .btn-send.hide { display: none; }
    .btn-img { background: #f0ebe0; color: #6b5b4f; position: relative; }
    .btn-img input { position: absolute; inset: 0; opacity: 0; cursor: pointer; }
    .preview { display: flex; align-items: center; gap: 5px; margin-bottom: 5px; padding: 3px 8px; background: #f8f5f0; border-radius: 5px; font-size: 11px; }
    .preview img { width: 28px; height: 28px; object-fit: cover; border-radius: 3px; }
    .preview .del { cursor: pointer; color: #e74c3c; margin-left: auto; font-size: 15px; }
  </style>
</head>
<body>
<div id="app">
  <div class="header">
    <h1>👨‍🍳 私厨 — AI 食谱助手</h1>
    <p>上传食材图片或输入清单，智能推荐菜谱</p>
  </div>
  <div class="chat" ref="wrap">
    <div class="chat-inner">
      <div v-if="msgs.length===0" class="empty">
        <div class="icon">🥬🥩🍳</div>
        <p>输入食材名称或上传食材照片</p>
      </div>
      <div v-for="m in msgs" :key="m.id" :class="['msg', m.role]">
        <div class="avt">{{ m.role==='user' ? '🙋' : '👨‍🍳' }}</div>
        <div class="bbl">
          <img v-if="m.img" :src="m.img">
          <!-- 用户消息保持纯文本；助手消息渲染 Markdown -->
          <span v-if="m.role==='user'">{{ m.text }}</span>
          <span v-else v-html="renderMd(m.text)"></span>
          <span v-if="m.streaming" class="cursor"></span>
          <span v-if="m.error" class="err">⚠ {{ m.error }}</span>
        </div>
      </div>
    </div>
  </div>
  <div class="input-bar">
    <div class="input-inner">
      <div class="thread-row">
        会话 <input v-model="tid" @change="loadHistory"> <a @click="clearHistory">清空</a>
      </div>
      <div class="preview" v-if="preview">
        <img :src="preview"> 图片已选 <span class="del" @click="preview='';file=null">&times;</span>
      </div>
      <div class="row">
        <button class="btn btn-img" :disabled="busy">
          📷 <input type="file" accept="image/*" @change="pickFile">
        </button>
        <textarea v-model="text" @keydown.enter.exact.prevent="send" placeholder="输入食材，如：鸡蛋、番茄、青椒..." :disabled="busy" rows="1"></textarea>
        <button class="btn btn-send" :class="{hide:busy}" @click="send" :disabled="busy||(!text.trim()&&!file)">发送</button>
        <button class="btn btn-stop" :class="{show:busy}" @click="stop">停止</button>
      </div>
    </div>
  </div>
</div>

<script>
const { createApp, ref, nextTick, onMounted } = Vue;
createApp({
  setup() {
    const msgs = ref([]);
    const text = ref('');
    const busy = ref(false);
    const tid = ref('user1');
    const file = ref(null);
    const preview = ref('');
    const wrap = ref(null);
    let abortCtrl = null;  // AbortController，用于取消请求

    /* ======== Markdown 渲染 ======== */
    function renderMd(src) {
      if (!src) return '';
      let html = src;
      // 转义 HTML 特殊字符（但保留已渲染的 HTML）
      html = html.replace(/&/g, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;');
      // 代码块 ```...```cpp
      html = html.replace(/```(\w*)\n?([\s\S]*?)```/g, (_, lang, code) =>
        '<pre><code>' + code.replace(/\n$/, '') + '</code></pre>');
      // 行内代码 `...`
      html = html.replace(/`([^`]+)`/g, '<code>$1</code>');
      // 粗体 **...**
      html = html.replace(/\*\*([^*]+)\*\*/g, '<strong>$1</strong>');
      // 斜体 *...*
      html = html.replace(/\*([^*]+)\*/g, '<em>$1</em>');
      // 图片 ![alt](url)
      html = html.replace(/!\[([^\]]*)\]\(([^)]+)\)/g, '<img src="$2" alt="$1">');
      // 链接 [text](url)
      html = html.replace(/\[([^\]]+)\]\(([^)]+)\)/g, '<a href="$2" target="_blank">$1</a>');
      // 标题 ### ...
      html = html.replace(/^### (.+)$/gm, '<h3>$1</h3>');
      html = html.replace(/^## (.+)$/gm, '<h2>$1</h2>');
      html = html.replace(/^# (.+)$/gm, '<h1>$1</h1>');
      // 无序列表 - 或 *
      html = html.replace(/^[\-\*] (.+)$/gm, '<li>$1</li>');
      // 有序列表 1. ...
      html = html.replace(/^\d+\. (.+)$/gm, '<li>$1</li>');
      // 包裹连续 <li> 为 <ul>
      html = html.replace(/((?:<li>.*<\/li>\n?)+)/g, '<ul>$1</ul>');
      // 水平线 ---
      html = html.replace(/^---+$/gm, '<hr>');
      // 引用 >
      html = html.replace(/^&gt; (.+)$/gm, '<blockquote>$1</blockquote>');
      // 换行
      html = html.replace(/\n\n/g, '</p><p>');
      html = html.replace(/\n/g, '<br>');
      // 包裹段落
      html = '<p>' + html + '</p>';
      // 清理空段落
      html = html.replace(/<p><\/p>/g, '');
      html = html.replace(/<p>(<h[123]>)<\/p>/g, '$1');
      html = html.replace(/(<\/h[123]>)<\/p>/g, '$1');
      html = html.replace(/<p>(<ul>)<\/p>/g, '$1');
      html = html.replace(/(<\/ul>)<\/p>/g, '$1');
      html = html.replace(/<p>(<pre>)<\/p>/g, '$1');
      html = html.replace(/(<\/pre>)<\/p>/g, '$1');
      html = html.replace(/<p>(<blockquote>)<\/p>/g, '$1');
      html = html.replace(/(<\/blockquote>)<\/p>/g, '$1');
      html = html.replace(/<p>(<hr>)<\/p>/g, '$1');
      return html;
    }

    /* ======== 滚动 ======== */
    function scroll() {
      nextTick(() => {
        const el = wrap.value;
        if (el) {
          // 流式输出时使用即时滚动，确保始终看到最新内容
          el.scrollTop = el.scrollHeight;
        }
      });
    }

    function mid() { return 'm_'+Date.now()+'_'+Math.random().toString(36).slice(2,6); }

    /* ======== 历史消息 ======== */
    async function loadHistory() {
      try {
        const r = await fetch('/api/v1/chat/messages?thread_id='+tid.value);
        if (r.ok) {
          const d = await r.json();
          msgs.value = (d.messages||[]).map(m => ({
            id: mid(),
            role: m.role,
            text: m.content,
            img: m.imageUrl
          }));
          scroll();
        }
      } catch(e) { /* 静默失败 */ }
    }

    async function clearHistory() {
      try {
        await fetch('/api/v1/chat/messages?thread_id='+tid.value, { method:'DELETE' });
        msgs.value = [];
      } catch(e) { /* 静默失败 */ }
    }

    /* ======== 图片 ======== */
    function pickFile(e) {
      const f = e.target.files[0];
      if (!f) return;
      file.value = f;
      preview.value = URL.createObjectURL(f);
    }

    async function upload(f) {
      const ext = f.name.split('.').pop()||'jpg';
      const fn = Date.now()+'.'+ext;
      const p = await fetch('/api/v1/oss/presign?filename='+encodeURIComponent(fn));
      if (!p.ok) throw new Error('获取上传地址失败');
      const { uploadUrl, accessUrl, contentType } = await p.json();
      const u = await fetch(uploadUrl, {
        method:'PUT',
        body: f,
        headers: { 'Content-Type': contentType }
      });
      if (!u.ok) throw new Error('上传失败: '+u.status);
      return accessUrl;
    }

    /* ======== 流式发送 ======== */
    async function send() {
      const msg = text.value.trim();
      const f = file.value;
      if (!msg && !f) return;

      // 如果正在流式输出，先取消
      if (abortCtrl) abortCtrl.abort();

      busy.value = true;
      abortCtrl = new AbortController();

      // 上传图片
      let imgUrl = '';
      if (f) {
        try {
          imgUrl = await upload(f);
        } catch(e) {
          msgs.value.push({
            id: mid(), role:'assistant', text:'',
            error: '图片上传失败: '+e.message
          });
          busy.value = false;
          abortCtrl = null;
          return;
        }
      }

      // 添加用户消息
      msgs.value.push({
        id: mid(),
        role: 'user',
        text: msg || '上传了食材图片',
        img: imgUrl || undefined
      });

      // 添加 AI 占位消息（流式输出目标）
      // 注意：必须从数组中取引用，不能直接用 push 前的对象 — Vue 3 的 Proxy 响应式
      // 只拦截通过 Proxy 的访问，直接修改原始对象不会触发 DOM 更新
      msgs.value.push({ id: mid(), role: 'assistant', text: '', streaming: true });
      const ai = msgs.value[msgs.value.length - 1];  // 取响应式 Proxy 引用
      scroll();

      // 清空输入
      text.value = '';
      preview.value = '';
      file.value = null;
      const fi = document.querySelector('.btn-img input');
      if (fi) fi.value = '';

      try {
        const r = await fetch('/api/v1/chat', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({
            message: msg || '这是什么食材？能做什么菜？',
            image_url: imgUrl,
            thread_id: tid.value
          }),
          signal: abortCtrl.signal
        });

        if (!r.ok) {
          const e = await r.json().catch(() => ({}));
          throw new Error(e.detail || '请求失败: '+r.status);
        }

        // === 流式读取 SSE ===
        const reader = r.body.getReader();
        const dec = new TextDecoder();
        let buf = '';         // 缓冲区，存放未完整接收的数据

        while (true) {
          const { done, value } = await reader.read();
          if (done) break;

          // 解码新收到的数据，追加到缓冲区
          buf += dec.decode(value, { stream: true });

          // 按行解析（兼容 \r\n 和 \n）
          while (true) {
            const idx = buf.search(/\r?\n/);
            if (idx === -1) break;  // 没有完整行，等待更多数据

            let line = buf.slice(0, idx);
            // 跳过 \r
            if (line.endsWith('\r')) line = line.slice(0, -1);
            // 移除已处理的部分（包括换行符）
            buf = buf.slice(idx + (buf[idx]==='\r' && buf[idx+1]==='\n' ? 2 : 1));

            // 空行 = SSE 事件边界，跳过
            if (!line) continue;

            // 解析 SSE 字段
            if (line.startsWith('data: ')) {
              const data = line.slice(6);
              if (data === '[DONE]') break;
              try {
                // 服务端 data 是 JSON 编码的字符串
                ai.text += JSON.parse(data);
              } catch(e) {
                // 解析失败则原样追加
                ai.text += data;
              }
            }
            // 忽略其他 SSE 字段（event:, id:, retry: 等）
          }
          scroll();
        }
      } catch(e) {
        // AbortError 是用户主动取消，不需要报错
        if (e.name === 'AbortError') {
          if (ai.text) {
            ai.text += '\n\n_（已停止生成）_';
          } else {
            ai.text = '（已取消）';
          }
        } else {
          ai.text = ai.text || '请求失败';
          ai.error = e.message;
        }
      }

      ai.streaming = false;
      busy.value = false;
      abortCtrl = null;
      scroll();
    }

    /* ======== 停止生成 ======== */
    function stop() {
      if (abortCtrl) {
        abortCtrl.abort();
        abortCtrl = null;
      }
    }

    onMounted(() => loadHistory());

    return {
      msgs, text, busy, tid, file, preview, wrap,
      loadHistory, clearHistory, pickFile, send, stop, renderMd
    };
  }
}).mount('#app');
</script>
</body>
</html>

配置

DEEPSEEK_API_KEY=sk-aaba5e9b7a71.....
DEEPSEEK_BASE_URL=https://api.deepseek.com

DASHSCOPE_API_KEY=sk-6c70387ce0c....
DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

TAVILY_API_KEY=tvly-dev-1Zc7wm-VM9VPkJASBQ5f....

# langsmith
LANGSMITH_API_KEY=lsv2_pt_bcbc096f8....
LANGSMITH_TRACING=false
# project name
LANGSMITH_PROJECT=lc-course


OSS_ACCESS_KEY_ID=LTAI5t....
OSS_ACCESS_KEY_SECRET=HYu09HyTtjC9laa....
OSS_BUCKET=ceshi...