我们先配置一下环境

1. 安装anaconda，具体安装方法查看页面：Python环境Anaconda下载与安装教程（附安装包） 2025最新版详细图文安装教程 - 知乎

   清华镜像站：清华大学开源软件镜像站

   安装好后创建一个langchain的环境，建议使用python3.10的版本（避免安装插件时冲突），安装langchain相关插件5个

   如果插件安装不上，可以打开工具 anaconda prompt

   conda create -n langchain python=3.10
   conda activate langchain
   conda install -c conda-forge langchain-openai
   

2. 安装PyCharm，社区版即可，2025.2是最后一个社区版，下载地址

   安装方法PyCharm安装教程及基本使用（更新至2024年新版本），教你迈出学习python第一步-CSDN博客

3. 自备一个云ai大模型环境，比如通义千问 通义大模型_AI大模型_一站式大模型推理和部署服务-阿里云

一、LangChain介绍

LangChain是基于大语言模型的应用开发框架，提供工具、组件和接口，让开发者像搭积木一样创建复杂AI应用。

特点

• 数据感知：能与多种数据源连接，将外部数据作为知识库，弥补大模型数据陈旧、无法实时获取信息等缺陷
• 代理交互：应用可主动与环境互动，如自动规划行程、总结邮件内容等，增强用户体验和工作效率。
• 解决痛点：针对大模型内部数据过时、无法联网、Token限制、无法调用API等问题，LangChain提供解决方案。

应用场景

• 文档摘要应用
• 智能客服
• 编程助手
• 智能私人助理
• 学习助手

模型支持

• LangChain封装了LLM类、Chat类和Embedding类模型，提供标准化接口。

  • LLM模型：接收文本输入，返回文本输出，无状态，不记交互历史。适用于快速问答、文本摘要、信息提取、代码生成等任务。

  • Chat类模型：接收消息列表，返回AI消息，支持多轮对话和角色扮演。构建聊天机器人、角色扮演应用、多轮交互任务等。

  • Embedding类模型：将文本映射为高维向量，用于语义搜索和知识库问答。实现语义搜索，是RAG应用的基石。

• 与OpenAI、Google、阿里等主流供应商合作，集成多种大模型。

  • 常见参数：temperature、top_p、n、max_tokens

• 封装基于OpenAI API接口规范，兼容模型可无缝使用。

代码体验

• 模型支持基本用法

from dotenv import load_dotenv
from langchain_community.llms import Tongyi

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

# 初始化通义千问 LLM 模型
llm = Tongyi(
    model_name="qwen-turbo",
    temperature=0.7,
    extra_body={"enable_thinking": False}
)

try:
    print("正在向模型发送请求...")
    # 调用模型，传入问题
    response = llm.invoke("你是谁？")
    print("模型返回结果：")
    # 打印模型的返回结果
    print(response)
except Exception as e:
    print(f"调用时发生错误: {e}")

• 聊天消息类模型

from dotenv import load_dotenv
from langchain_community.llms import Tongyi
from langchain_core.messages import HumanMessage, SystemMessage

# 加载环境变量
load_dotenv()

# 初始化通义千问 Chat 模型
# 注意这里我们是从 chat_models 导入的
chat = Tongyi(
    model_name="qwen-turbo",
    temperature=0.7,
    extra_body={"enable_thinking": False}
)

# 构建一个包含角色信息的消息列表
messages = [
    SystemMessage(content="你是一个不爱说话的植物学家，回答问题时总是意简言赅，惜字如金。"),
    HumanMessage(content="什么是西瓜？")
]

try:
    # 将消息列表传递给模型
    response = chat.invoke(messages)
    # 打印模型的返回结果（这是一个AIMessage对象）
    print(response)
except Exception as e:
    print(f"调用时发生错误: {e}")

• OPenAI接口集成

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 加载环境变量
load_dotenv()

# 从环境变量中获取API地址和密钥
# 注意这里我们是从 os 模块中获取了
api_base = os.getenv("DASHSCOPE_API_BASE")
api_key = os.getenv("DASHSCOPE_API_KEY")

# 使用 ChatOpenAI 类进行初始化
chat = ChatOpenAI(
    base_url=api_base,
    api_key=api_key,
    model_name="qwen-turbo",
    temperature=0.7,
    top_p=1,
    n=3,
    max_tokens=1024,
    streaming=False,
    stop_sequences=["\n"],
    logit_bias= {},
    frequency_penalty=0, # 惩罚高频词
    presence_penalty=0,# 惩罚出现过的词
    extra_body={"enable_thinking": False}
)

messages = [
    SystemMessage(content="你是一个不爱说话的植物学家，回答问题时总是意简言赅，惜字如金。"),
    HumanMessage(content="什么是西瓜？")
]

try:
    response = chat.invoke(messages)
    # 打印返回消息中的 content 属性
    print(response.content)
except Exception as e:
    print(f"调用时发生错误: {e}")

二、工具调用

1. 工具调用基础概念

• 打破数据限制：大模型通过工具调用可获取实时信息，如调用天气API回答即时天气问题，还能与私有数据交互，成为懂业务的专属助理，甚至在真实世界执行操作，如发送邮件、预订机票等，从“聊天者”蜕变为“行动派”，极大地扩展了应用边界。
• 工作原理：工具调用过程可类比为CEO与助理的分工。大模型接收指令后，判断是否需要外部工具，生成调用指令，由应用程序执行，再将结果反馈给大模型，最终以自然语言形式呈现给用户。整个流程包括提供工具蓝图、大模型推理规划、生成调用指令、代码执行、结果反馈及大模型生成答复等步骤。
• 工具调用是一种单轮交互过程，通常是线性的“一问一答”形式，大模型根据任务需求选择工具并生成调用指令，由代码执行并返回结果。
• 工具调用与Agent的区别：Agent是一种利用工具调用能力的架构，更像自动驾驶系统。它通过“思考 - 行动 - 观察”的循环结构，不断重复调用工具，直到获取足够信息回答问题。工具调用是构建Agent的基础模块，学会单次调用才能构建复杂的Agent系统。

2. 工具调用流程说明

• 用户提问："3*12是多少？"
• 系统处理步骤： ① 展示工具列表及说明书（名称/功能/参数） ② 大模型选择工具并生成结构化JSON指令 ③ 代码执行：解析JSON→调用函数/API ④ 返回工具执行结果（如"36"）并包装为特定消息格式 ⑤ 大模型生成最终自然语言回复
• 大模型像CEO在办公室：阅读请求→撰写JSON指令→代码处理（如财务部API查询数据）→返回结果

3. 基于提示词的工具调用

常见方法

• 格式化输出要求：开发者在提示词中明确要求模型按特定格式（如JSON）返回结果，使程序能轻松解析并实现自动化操作。
• 少样本学习示例：通过在提示词中提供少量标准答案示例，让模型学习如何从用户输入中提取关键信息并按预设格式输出，利用模型的上下文学习能力，使输出更贴近预期。
• 严格的指令和约束：在提示词中增加详尽的约束条件，如“仅输出JSON，不要附加任何解释性文字”，防止模型自由发挥，降低生成无关废话的风险，确保结果可被代码顺利解析。
• 后处理步骤：即便提示词设计完美，模型输出仍可能有偏差，开发者在应用程序层面增加后处理步骤，通过正则表达式提取、JSON合法性检查等手段，过滤整理输出，保证数据格式一致性。

存在的弊端

• 稳定性不足：模型输出依赖“临场发挥”，可能附加额外文本或生成格式错误的JSON，导致程序解析失败，难以保证输出一致性。
• 提示词设计复杂：为覆盖各种情况，提示词需写得冗长且难以维护，微小改动可能影响结果格式和内容。
• 错误处理困难：遇到用户不按常理出牌的输入或边缘案例时，基于提示词的方案难以自动纠错，需编写额外后处理逻辑，增加开发和维护成本。
• 扩展性差：支持的函数数量增加时，提示词需不断更新扩充，且受模型Token长度限制，扩展和调试变得繁琐低效。

提示词举例

你是一个多功能助手，能够根据用户意图选择并返回合适的工具调用信息。下面是你可以调用的工具列表：
1. 查询天气工具：
	工具名称：get_weather
	参数：{
		"location": 城市名称,
		"date"：查询日期(如“今天”，“2025-08-20”)
	}
	用途：用于查询某个城市在指定日期的天气情况
2. 获取新闻工具：
	工具名称：get_news
	参数：{
		"topic": 新闻主题(如“体育”，“政治”)
	}
	用途：用于获取与特定主题相关的最新新闻
3. 讲笑话工具：
	工具名称：get_joke
	参数：无
	用途：用于返回一个笑话
	
当用户发送消息时，请先分析用户意图，并返回一个json格式的字符串，仅包含要调用的工具名称及对应参数，格式如下（主意：只返回json，不附加任何其他文字）：
{
	"function"： "<工具名称>",
	"parameters"：{ ... }
}	

用户输入	标准化响应
北京明天天气？	{"function":"get_weather","parameters":{"location":"北京","date":"明天"}}
最近体育新闻	{"function":"get_news","parameters":{"topic":"体育"}}
讲个笑话	{"function":"get_joke","parameters":}

4. 大模型工具调用实现原理

• 内置函数描述与注册：开发者将可调用函数的名称、用途、参数等以结构化数据格式（如JSON Schema）提供给模型，作为模型的“官方参考手册”，模型据此判断是否调用及如何调用某个函数。
• 结构化输出生成：新一代模型在训练时接触大量结构化数据和格式化调用示例，能生成严格符合预定义格式的输出，准确判断输出普通文本还是结构化工具调用指令，减少对复杂提示词和后处理逻辑的依赖。
• 自动推理与调用选择：大模型凭借强大的自然语言理解能力，精准识别用户意图，自动从工具库中挑选合适工具调用，并从用户输入中提取信息填充参数，生成规范的工具调用请求。

5. LangChain工具调用实现全流程

简单的计算器应用===>两个工具：一个加法工具、一个乘法工具

3乘以12等于多少？12加上36是多少？

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage

# 从环境变量中获取API地址和密钥
# 注意这里我们是从 os 模块中获取了
load_dotenv()
api_base = os.getenv("DASHSCOPE_API_BASE")
api_key = os.getenv("DASHSCOPE_API_KEY")

# 定义工具

## 标准Python函数
def multiply(a: int, b: int) -> int:
    """
    用于计算两个整数的乘积。

    Args:
        a: 第一个整数。
        b: 第二个整数。
    """
    return a * b
def add(a: int, b: int) -> int:
    """
    用于计算两个整数的和。

    Args:
        a: 第一个整数。
        b: 第二个整数。
    """
    return a + b

# 将我们定义好的工具放入一个列表中，方便后续使用。
tools = [multiply, add]

# 实例化模型并绑定工具
llm = ChatOpenAI(
    base_url=api_base,
    api_key=api_key,
    model_name="qwen-plus",
    extra_body={"enable_thinking": False, "parallel_tool_calls": True})

# 使用.bind_tools() 方法将我们的工具列表“绑定”到模型上。
# 这个方法会处理所有繁琐的工作，将Python工具转换成模型API所需的格式。
# 之后，每次调用 llm_with_tools 时，这些工具的信息都会随请求一起发送给模型。
llm_with_tools = llm.bind_tools(tools)

# 调用模型，获取工具调用请求

# 我们将整个对话过程保存在一个消息列表中。
# 对话从一个HumanMessage开始。
query = "3乘以12等于多少？12加上36是多少？"
messages = [HumanMessage(content=query)]

print(f"--- 用户问题 ---\n{query}\n")

# 调用绑定了工具的模型。
# 此时，模型不会直接回答问题，而是会分析问题并决定调用哪些工具。
ai_msg = llm_with_tools.invoke(messages)

# 将模型的响应（一个AIMessage）也添加到消息历史中。
messages.append(ai_msg)

# 打印模型返回的AIMessage，重点关注其.tool_calls 属性。
# .tool_calls 是一个列表，包含了模型希望我们调用的所有工具的信息。
print(f"--- 模型的首次响应 (AIMessage) ---\n{ai_msg}\n")
print(f"--- 模型请求的工具调用 (tool_calls) ---\n{ai_msg.tool_calls}\n")

# 执行工具并且创建工具消息

# 现在，我们需要遍历模型请求的每一个工具调用，并实际执行它们。
# 为了方便地从名称映射到工具本身，我们创建一个字典。
tool_map = {
    "multiply": multiply,
    "add": add
}

# 遍历 ai_msg.tool_calls 列表
for tool_call in ai_msg.tool_calls:
    # 根据模型提供的工具名称，从我们的映射中选择正确的工具。
    selected_tool = tool_map.get(tool_call["name"].lower())

    if selected_tool:
        # 执行工具
        # tool_call["args"] 是一个字典，包含了模型为我们提取的参数。
        # 我们可以使用 ** 操作符将其解包并传递给工具。
        tool_output = selected_tool(**tool_call["args"]) # **是python的解包语法
        print(f"--- 执行工具 '{tool_call['name']}' ---")
        print(f"参数: {tool_call['args']}")
        print(f"输出: {tool_output}\n")

        # 将工具的执行结果包装成一个 ToolMessage。
        # 重要的是，必须包含 tool_call_id，这样模型才能将结果与它之前的请求对应起来。
        messages.append(ToolMessage(content=str(tool_output), tool_call_id=tool_call["id"]))
    else:
        print(f"警告：模型请求了一个未知的工具 '{tool_call['name']}'")

# 再次调用模型，获取最终的自然语言答案
# 现在，我们的 messages 列表包含了：
# 1. 用户的原始问题 (HumanMessage)
# 2. 模型请求调用工具的响应 (AIMessage with tool_calls)
# 3. 每个工具执行后的结果 (ToolMessage)

print(f"--- 包含工具结果的完整消息历史 ---\n{messages}\n")

# 我们用这个包含了工具执行结果的完整消息历史再次调用模型。
# 这次，模型拥有了回答原始问题所需的所有信息。
final_response = llm.invoke(messages)

print(f"--- 模型的最终回答 ---\n{final_response.content}\n")

6. 工具定义的多种方式

• 标准python函数：上面的例子就是使用python函数定义的工具

• 使用tool装饰器：

  from langchain_core.tools import tool
  
  @tool
  def multiply(first_number: int, second_number: int) -> int:
      """这个工具用于计算两个整数的乘积。"""
      return first_number * second_number
  
  # 示例2：自定义工具名称
  # 有时候，工具名可能不够清晰，或者我们想给大模型一个更具体的指令
  # 我们可以给 @tool 装饰器传递一个字符串参数，来指定工具的名称
  @tool("special_add_tool")
  def add(a: int, b: int) -> int:
      """这个工具用于计算两个整数的和。"""
      return a + b
  
  # 执行的时候使用
  tool_output = selected_tool.invoke(tool_call["args"])
  

• 使用Pydantic模型

  from pydantic import BaseModel, Field
  
  ## 使用Pydantic模型：数据验证
  ## 定义数据模板
  class MultiplyInput(BaseModel):
      """用于 multiply 工具的输入参数模型。"""
  
      # 为模板定义字段（参数）
      # 格式是：参数名: 类型 = Field(..., description="...")
  
      # 'query' 是一个必需的字符串参数。
      # Field(..., description="...") 中的... 表示这个字段是必需的。
      # description 的内容对大模型至关重要，它解释了这个参数的用途。
      a: int = Field(..., description="第一个整数。")
      b: int = Field(..., description="第二个整数。")
  
  # 使用 @tool 装饰器来定义工具
  # docstring 成为工具的整体描述
  # args_schema 指向我们刚刚定义的 Pydantic 模型
  @tool(args_schema=MultiplyInput)
  def multiply(a: int, b: int) -> int:
      """用于计算两个整数的乘积。"""
      return a * b
  
  # 为 'add' 工具定义一个 Pydantic 输入模型
  class AddInput(BaseModel):
      """用于 add 工具的输入参数模型。"""
      a: int = Field(..., description="第一个整数。")
      b: int = Field(..., description="第二个整数。")
  
  @tool(args_schema=AddInput)
  def add(a: int, b: int) -> int:
      """用于计算两个整数的和。"""
      return a + b
  
  # 工具字典改成动态创建
  tool_map = {tool.name: tool for tool in tools}
  

• 使用TypedDict类型字典（LangChain 的最新版本不再直接支持 Python 原生的 TypedDict，这里不说了）

方法	实现方法	简洁性	模型控制力	内置验证	依赖	最适合场景
标准函数	def my_func(...)	高	低	否	无	快速原型开发，简单工具
@tool装饰器	@tool def my_func(...)	高	中	否	langchain-core	需要自定义工具名称或简单参数描述
Pydantic模型	class MyInput(BaseModel)	中	高	是	pydantic	生产级应用，需要严格数据验证的复杂API

7. 工具描述最佳实践

• 使用清洗、面向行动的命名：
  • 差：data
  • 好：get_user_profile
  • 更好：fetch_user_profile_by_email
• 在描述中明确工具何时使用：
  • 差：搜索工具
  • 好：执行网络搜索
  • 更好：当用户提出关于xxx的问题时，使用此工具进行网络搜索以获取最新答案
• 为参数提供详尽的描述和示例
  • 差：city:str
  • 好：city:str=Field(description="用户想要查询天气的城市名称")
  • 更好：city:str=Field(description="用户想要查询天气的城市名称，例如'beijing','shenzhen',应为拼音")
• 保存工具功能的单一和专注
  • 差：万能工具
  • 好：三个独立工具：get_weather \ ...
  • 单一职责的工具更容易被模型理解和正常调用

三、LangChain的几个技巧

1. 流式传输

• 延迟问题：大语言模型响应时间长，调用外部API等操作会进一步放大延迟，导致用户体验差。
• 流式处理优势：流式处理可提前展示部分输出，让用户实时了解进度，提升应用灵敏度和交互性。
• 实现：流式调用返回的是AIMessageChunk对象的迭代器，包含tool_call_chunks和tool_calls。 tool_call_chunks是原始字符串片段，适合实时展示；tool_calls是解析后的工具调用版本，供程序使用。
• 代码示例：使用stream()方法代替invoke()，通过循环接收数据块，LangChain自动聚合数据块，最终得到完整的工具调用信息。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool

# 从环境变量中获取API地址和密钥
load_dotenv()
api_base = os.getenv("DASHSCOPE_API_BASE")
api_key = os.getenv("DASHSCOPE_API_KEY")

# 定义一个简单的乘法工具
@tool
def multiply(a: int, b: int) -> int:
    """用于计算两个整数的乘积。"""
    print(f"服务器端：正在执行乘法 {a} * {b}...")
    result = a * b
    print(f"服务器端：乘法结果为 {result}")
    return result

# 初始化大模型
llm = ChatOpenAI(
    base_url=api_base,
    api_key=api_key,
    model_name="qwen-plus",
    streaming=True,
    extra_body={"enable_thinking": False, "parallel_tool_calls": True})

# 将工具绑定到模型上
tools = [multiply]
llm_with_tools = llm.bind_tools(tools)

# 我们的问题
message = [HumanMessage(content="计算一下 12 乘以 13 等于多少？")]

# 开始流式处理
print("客户端：向模型发起请求...")
# 使用 stream() 方法获取异步迭代器
stream = llm_with_tools.stream(message)

final_chunk = None
for chunk in stream:
    # chunk 是一个 AIMessageChunk 对象
    # 打印每个块的 tool_call_chunks 以观察底层过程
    if chunk.tool_call_chunks:
        # print(f"客户端：收到完整数据块 -> {chunk}")
        print(f"客户端：收到工具调用块 -> {chunk.tool_call_chunks}")

    # 聚合 AIMessageChunk
    if final_chunk is None:
        final_chunk = chunk
    else:
        # AIMessageChunk 的 '+' 操作符会自动处理 tool_call_chunks 的聚合
        final_chunk += chunk

print("\n--------------------")
print(f"客户端：流接收完毕，最终解析到的完整工具调用 -> {final_chunk.tool_calls}")
print("--------------------")

# 开始执行工具
tool_map = {tool.name: tool for tool in tools}
tool_outputs = []
if final_chunk.tool_calls:
    for call in final_chunk.tool_calls:
        # call 是一个字典，包含了 name, args, id
        print(f"客户端：准备执行工具 -> {call['name']}，参数 -> {call['args']}")
        # 根据模型提供的工具名称，从我们的映射中选择正确的函数。
        selected_tool = tool_map.get(call["name"].lower())

        if selected_tool:
            # 执行工具
            output = selected_tool.invoke(call["args"])
            # 将结果封装成 ToolMessage
            tool_outputs.append(
                ToolMessage(content=str(output), tool_call_id=call["id"])
            )

# 构造完整的对话历史
# 顺序必须是：HumanMessage -> AIMessage(with tool_calls) -> ToolMessage
# 将模型的第一轮回复（包含工具调用请求）添加到历史记录
message.append(final_chunk)
# 将工具执行的结果添加到历史记录
message.extend(tool_outputs)
final_stream = llm_with_tools.stream(message)

print("客户端：正在接收模型的最终回复...")
final_answer = ""
for chunk in final_stream:
    content = chunk.content or ""
    print(content, end="", flush=True)
    final_answer += content

print("\n\n客户端：对话结束。")

# 以下是输出内容
客户端：向模型发起请求...
客户端：收到工具调用块 -> [{'name': 'multiply', 'args': '{"a": 1', 'id': 'call_bd97f2c8a2624f81bbc2b3', 'index': 0, 'type': 'tool_call_chunk'}]
客户端：收到工具调用块 -> [{'name': None, 'args': '2, "b":', 'id': '', 'index': 0, 'type': 'tool_call_chunk'}]
客户端：收到工具调用块 -> [{'name': None, 'args': ' 13}', 'id': '', 'index': 0, 'type': 'tool_call_chunk'}]

--------------------
客户端：流接收完毕，最终解析到的完整工具调用 -> [{'name': 'multiply', 'args': {'a': 12, 'b': 13}, 'id': 'call_bd97f2c8a2624f81bbc2b3', 'type': 'tool_call'}]
--------------------
客户端：准备执行工具 -> multiply，参数 -> {'a': 12, 'b': 13}
服务器端：正在执行乘法 12 * 13...
服务器端：乘法结果为 156
客户端：正在接收模型的最终回复...
12 乘以 13 等于 156。

客户端：对话结束。

2. 运行时值注入

• 参数级注入：使用InjectedToolArg标记工具函数参数，使该参数对大模型不可见，仅在运行时由应用程序注入。

import json
from typing import List
from typing_extensions import Annotated
from langchain_core.tools import tool, InjectedToolArg

@tool
def update_favorite_pets(
    pets: List[str],
    user_id: Annotated[str, InjectedToolArg],
):
    """为指定用户更新最喜爱的宠物列表。"""
    # 假设这是函数的实现
    print(f"为用户 {user_id} 更新宠物列表为: {pets}")
    return {"status": "success", "user_id": user_id, "pets": pets}

# 看看完整的输入说明书长啥样
print("## 完整输入 Schema (供内部使用) ##")
print(json.dumps(update_favorite_pets.get_input_schema().model_json_schema(), indent=2, ensure_ascii=False))

print("\n" + "="*50 + "\n")

# 再看看给大模型看的调用说明书长啥样
print("## LLM 调用 Schema (仅包含 LLM 需要提供的参数) ##")
print(json.dumps(update_favorite_pets.tool_call_schema.model_json_schema(), indent=2, ensure_ascii=False))

• 可运行配置：使用RunnableConfig传递全局配置信息，如API密钥等，自动传导至调用链中的所有组件。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage
# 导入可运行配置对象
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool

load_dotenv()
api_base = os.getenv("DASHSCOPE_API_BASE")
api_key = os.getenv("DASHSCOPE_API_KEY")

# 假设这是一个模拟的外部天气API客户端
def query_weather_api(location: str, api_key: str) -> str:
    if not api_key:
        return f"错误：无效API密钥"
    return f"成功使用API密钥 '{api_key[:5]}...' 查询到 '{location}' 的天气为多云。"

@tool
def get_weather(location: str, config: RunnableConfig) -> str:
    """
    获取指定地点的当前天气。
    """
    print(f"工具内部收到的RunnableConfig: {config}")

    # 从 'configurable' 字典里安全地拿出我们的API密钥
    api_key_from_config = config.get("configurable", {}).get("weather_api_key")

    # 调用真正的API服务
    return query_weather_api(location, api_key_from_config)

# 初始化模型并绑定工具
llm = ChatOpenAI(
    base_url=api_base,
    api_key=api_key,
    model_name="qwen-plus",
    extra_body={"enable_thinking": False, "parallel_tool_calls": True})
llm_with_tool = llm.bind_tools([get_weather])

# 模拟一个运行时环境，在这里我们提供一个临时的API密钥
runtime_api_key = "这是一个临时的密钥_for_this_request_12345"
runtime_config = {
    "configurable": {
        "weather_api_key": runtime_api_key,
        "user_id": "user_gamma", # 我们也可以传递其他任何需要的信息
        "session_id": "session_xyz"
    }
}

messages = [HumanMessage(content="北京今天天气怎么样？")]
# 调用模型，并把我们准备好的配置字典传进去
response = llm_with_tool.invoke(
    messages,
    config=runtime_config
)

print("\n大模型的响应（包含工具调用请求）:")
print(response.tool_calls)

# 在实际应用中，我们接下来需要执行工具并把结果返回给模型
# 这里为了演示方便，我们就直接调用一下工具看看效果
if response.tool_calls:
    tool_call = response.tool_calls[0]
    # 注意，因为我们是手动调用工具，所以也需要把config传进去，以便工具能访问到它
    tool_output = get_weather.invoke(tool_call['args'], config=runtime_config)
    print(f"\n工具 '{tool_call['name']}' 的执行输出: {tool_output}")

再来看一个示例

import os
from dotenv import load_dotenv
from typing import List, Dict
from typing_extensions import Annotated

from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import InjectedToolArg, tool
from langchain_openai import ChatOpenAI

load_dotenv()
api_base = os.getenv("DASHSCOPE_API_BASE")
api_key = os.getenv("DASHSCOPE_API_KEY")

# 使用一个字典来模拟一个多租户的数据库
user_tasks_db: Dict[str, List[str]] = {
    "user_a": ["完成报告"],
    "user_b": ["预约牙医"],
}

@tool
def add_task(task: str, user_id: Annotated[str, InjectedToolArg]) -> str:
    """给用户的待办事项列表里加个新任务。"""
    # setdefault 会尝试获取 key 对应的 value
    # 如果 key 不存在，则会先为它设置一个默认值（这里是空列表[]），然后再返回这个默认值。
    # 这样一来，之后的 .append() 操作就可以安全地在这个列表上进行了。
    user_tasks_db.setdefault(user_id, []).append(task)
    return f"任务 '{task}' 已经成功为用户 '{user_id}' 添加。"

@tool
def list_tasks(user_id: Annotated[str, InjectedToolArg]) -> List[str]:
    """列出某个用户的所有待办事项。"""
    # dict.get() 方法在key不存在时，可以安全地返回一个默认值（这里是空列表）。
    return user_tasks_db.get(user_id, [])

tools = [add_task, list_tasks]


# 1. 初始化模型并把工具“告诉”它
llm = ChatOpenAI(
    base_url=api_base,
    api_key=api_key,
    model_name="qwen-plus",
    extra_body={"enable_thinking": False})
llm_with_tools = llm.bind_tools(tools)

# 2. 模拟一个来自用户的请求
# 假设我们通过会话管理，得知当前登录的用户是 "user_b"
current_user_id = "user_b"
messages = [HumanMessage(content="帮我添加一个任务：交电费")]

# 3. 调用大模型，让它决定应该使用哪个工具
# 大模型只会根据用户的提问和它能看到的工具说明书来做决策
ai_msg = llm_with_tools.invoke(messages)

print("--- 大模型的原始输出 ---")
print(ai_msg)
print("\n--- 大模型生成的工具调用请求（注意，完全不包含user_id） ---")
print(ai_msg.tool_calls)

# 4. 手动注入运行时值并执行工具
if ai_msg.tool_calls:
    # 为了演示方便，我们假设只处理第一个工具调用
    tool_call = ai_msg.tool_calls[0]

    # 准备一个工具的映射，方便我们根据名字找到对应的工具函数
    tool_map = {tool.name: tool for tool in tools}

    # 找到大模型决定要调用的那个工具
    selected_tool = tool_map[tool_call["name"]]

    # 这是关键的一步：注入运行时需要的值！
    # 我们把大模型生成的参数和我们在运行时才知道的 user_id 合并起来
    tool_args = tool_call["args"].copy()  # 复制一份以防万一
    tool_args["user_id"] = current_user_id

    print(f"\n--- 注入 user_id 后的完整参数 ---")
    print(tool_args)

    # 5. 用这个包含了所有参数的完整字典来执行工具
    result = selected_tool.invoke(tool_args)

    print(f"\n--- 工具执行结果 ---")
    print(result)

    print(f"\n--- 数据库更新后的状态 ---")
    print(user_tasks_db)

3. 提示模板

• 定义：提示模板是包含固定文本和动态占位符的字符串，用于生成标准化提示。
• 优势：提高一致性、效率，改善模型性能，增强灵活性与可扩展性。

from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate

chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的AI助手，你的名字叫{name}。"),
    ("human", "你好，最近怎么样？"),
    ("ai", "我很好，谢谢！"),
    ("human", "{user_input}"),
    HumanMessage(content="你好")
])

# 使用 invoke 方法来填充模板中的变量
messages = chat_template.invoke({
    "name": "小助手",
    "user_input": "你的名字是什么？"
})

# invoke 返回的是一个 PromptValue 对象
print("--- PromptValue 对象 ---")
print(messages)

# 我们可以使用 to_messages() 方法来获取标准的消息对象列表
print("\n--- 消息对象列表 ---")
print(messages.to_messages())

• 字符串提示模版：使用构造函数或类方法from_template创建，可通过invoke方法填充变量并获取PromptValue对象。

from langchain_core.prompts import PromptTemplate

# # 使用构造函数创建模板，需要我们手动指定 input_variables
prompt_template = PromptTemplate(
    input_variables=["country", "cuisine"],
    template="为一家在 {country} 经营 {cuisine} 菜系的餐厅推荐一个名字。",
    # validate_template=False,
)
#
# # 使用 format 方法来格式化模板
# formatted_prompt = prompt_template.format(country="中国", cuisine="川菜")
# print(formatted_prompt)

# 使用 from_template 这个类方法来创建，LangChain 会自动从模板字符串中推断出输入变量
prompt_template = PromptTemplate.from_template(
    "告诉我一个关于{topic}的笑话。"
)

# 我们可以查看一下自动推断出的输入变量
print(f"输入变量: {prompt_template.input_variables}")

# 使用 invoke 方法来填充模板并获取一个 PromptValue 对象
prompt_value = prompt_template.invoke({"topic": "程序员"})

# 将 PromptValue 对象转换为字符串并打印出来
print(prompt_value.to_string())

• 聊天提示模板：通过ChatPromptTemplate.from_messages定义多角色对话，支持动态插入对话历史记录，使用MessagesPlaceholder实现。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

chat_template_with_history = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的客服机器人。"),
    # 在这里为我们的对话历史创建一个占位符
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}")
])

# 模拟一段已经存在的对话历史
history = [
    HumanMessage(content="我的订单发货了吗？"),
    AIMessage(content="请提供您的订单号，我帮您查询。")
]

# 调用 invoke 时，同时传入历史记录和当前的问题
final_messages = chat_template_with_history.invoke({
    "chat_history": history,
    "question": "我的订单号是12345。"
})

print(final_messages.to_messages())

• 少样本提示模板

  • 少样本提示是一种通过少量示例指导模型的技术，无需微调模型，即可提升其在特定任务上的表现，适用于格式化输出、风格模仿和复杂推理等场景。
  • 该技术无需修改模型结构或进行耗时的微调，只需通过少量示例即可让模型快速理解任务逻辑，显著提升模型在垂直任务上的表现力。
  • 模板构成：少样本提示模板包含示例列表、示例提示、前缀、后缀、输入变量和示例分割符等关键部分，用于生成结构清晰的长字符串提示。

  from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
  
  # 1. 定义要提供给模型的示例
  # 这就是我们为模型准备的“教科书”
  examples = [
      {"word": "高兴", "antonym": "悲伤"},
      {"word": "高", "antonym": "矮"},
      {"word": "快", "antonym": "慢"},
  ]
  
  # 2. 创建一个模板，用于格式化每一个示例
  # 这个模板定义了每个示例如何向模型进行自我介绍
  example_template = "词语: {word}\n反义词: {antonym}"
  example_prompt = PromptTemplate(
      input_variables=["word", "antonym"],
      template=example_template
  )
  
  # 3. 创建 FewShotPromptTemplate 实例
  # 这是一个组装器，它会将所有部分整合起来
  few_shot_prompt = FewShotPromptTemplate(
      examples=examples,
      example_prompt=example_prompt,
      prefix="请给出每个输入词语的反义词。",
      suffix="词语: {input}\n反义词:",
      input_variables=["input"],
      example_separator="\n\n" # 使用两个换行符分隔示例，保持清晰
  )
  
  # 4. 格式化并查看最终生成的完整提示
  # 让我们看看当用户输入“强”时，最终的提示长什么样
  final_prompt = few_shot_prompt.invoke({"input": "强"})
  print(final_prompt.to_string())
  

  通过代码示例展示了如何构建一个智能生成反义词的少样本提示，体现了少样本提示模板在文本补全模型中的具体实现和应用方法。

  • 聊天模型的少样本提示：为解决文本补全模型与聊天模型输入格式不匹配的问题，LangChain提供了少样本聊天提示模板，专门用于生成符合聊天模型要求的消息对象列表。

  from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
  
  # 1. 定义聊天格式的示例
  # 注意，这里的结构反映了人与AI的一问一答
  chat_examples = [
      {"input": "2+2", "output": "4"},
      {"input": "2*3", "output": "6"},
  ]
  
  # 2. 创建一个 ChatPromptTemplate 来格式化每个示例
  # 每个示例字典将被转换为一对 HumanMessage 和 AIMessage
  example_chat_prompt = ChatPromptTemplate.from_messages([
      ("human", "{input}"),
      ("ai", "{output}"),
  ])
  
  # 3. 创建 FewShotChatMessagePromptTemplate 实例
  # 它会使用上一步定义的 example_chat_prompt 来处理 chat_examples
  few_shot_chat_prompt = FewShotChatMessagePromptTemplate(
      examples=chat_examples,
      example_prompt=example_chat_prompt,
  )
  
  # 4. 将其嵌入到一个最终的聊天模板中，形成完整的对话上下文
  final_prompt_template = ChatPromptTemplate.from_messages([
      ("system", "你是一个强大的数学计算器。"),
      # 少样本示例将以消息列表的形式被无缝插入到这里
      few_shot_chat_prompt,
      # 用户的实际输入将作为对话的最后一环
      ("human", "{input}"),
  ])
  
  # 5. 调用并查看生成的消息列表
  # 当用户输入“5*5”时，我们来看看最终生成了怎样的消息序列
  final_messages = final_prompt_template.invoke({"input": "5*5"})
  print(final_messages.to_messages())
  

  通过代码示例展示了如何构建一个数学计算器的少样本聊天提示，体现了少样本聊天提示模板在聊天模型中的具体实现和应用方法。

• 管道提示模板

  • 定义与作用：管道提示模板允许将复杂提示分解为多个逻辑独立且可复用的子模板，通过流水线方式串联起来，最终组合成完整连贯的提示。
  • 核心优势：该模板具有复用性、可维护性和清晰度等优势，降低了心智负担，便于独立理解和测试，提升了提示工程的效率和质量。
  • 管道提示模板的构成
    • 管道提示列表：管道提示列表由元组（name, prompt_template）组成，定义了管道的各个阶段，每个提示模板都会被格式化并生成结果供后续模板使用。
    • 最终提示：最终提示是管道流水线的最后一个模板，接收管道提示列表的生成结果作为输入变量，并将它们组合、格式化成最终的提示输出。
    • 应用场景：管道提示模板适用于需要将复杂提示分解为多个逻辑部分的场景，如邮件生成提示等，可提高提示的可维护性和复用性。

from langchain_core.prompts import PromptTemplate, PipelinePromptTemplate

# 1. 定义各个部分的子模板，每个模板各司其职
# 负责定义角色
introduction_template = PromptTemplate.from_template("你正在扮演 {person}。")
# 负责提供一个对话示例
example_template = PromptTemplate.from_template("这是一个互动示例：\n问: {example_q}\n答: {example_a}")
# 负责引出真正的问题
start_template = PromptTemplate.from_template("现在，请真实地回答这个问题：\n问: {input}\n答:")

# 2. 定义最终的组合模板，它像一个骨架，包含了三个占位符
# 这三个占位符将分别接收上面三个子模板的输出
full_template = PromptTemplate.from_template(
    """{introduction}
        {example}
        {start}"""
)

# 3. 创建管道提示模板，将子模板与最终模板连接起来
pipeline_prompt = PipelinePromptTemplate(
    final_prompt=full_template,
    pipeline_prompts=[
        ("introduction", introduction_template),
        ("example", example_template),
        ("start", start_template)
    ]
)

# 4. 查看管道模板自动推断出的所有输入变量
# LangChain 会智能地分析整个管道，找出所有需要的“原料”
print(f"管道模板的输入变量: {sorted(pipeline_prompt.input_variables)}")

# 5. 调用并格式化，一次性传入所有子模板需要的变量
final_prompt_value = pipeline_prompt.invoke({
    "person": "埃隆·马斯克",
    "example_q": "你最喜欢的车是什么？",
    "example_a": "特斯拉",
    "input": "你最喜欢的社交媒体网站是什么？"
})

print("\n--- 生成的最终提示 ---")
print(final_prompt_value.to_string())

通过代码示例展示了如何将角色介绍、互动示例和实际问题三个独立模板组合成一个完整的角色扮演提示，体现了管道提示模板的强大功能和应用价值。

4. 预填充变量

• 定义与作用：预填充允许预先填充模板中的部分变量，创建新的、接口更简洁的模板，解决了复杂链式调用中变量管理的问题，提升了代码的模块化程度。
• 核心优势：预填充技术类似于函数式编程中的偏函数应用，能够将提示的定义过程与变量的获取过程解耦，使代码更加清晰和易于维护。
• 使用静态值预填充
  • 实现方法：当提前知道某个变量的固定值时，可以使用预填充将其固化到模板中，通过.partial()方法返回一个新的模板实例，简化了调用方的接口。
  • 应用场景：适用于问候语、日期等固定值的场景，通过预填充可减少变量传递的负担，使代码更加简洁高效。

from datetime import datetime
from langchain_core.prompts import PromptTemplate

# 定义一个函数，用于获取当前格式化的时间
def get_current_datetime():
    return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

# 在创建模板时，直接通过 partial_variables 参数传入函数本身
prompt = PromptTemplate(
    template="报告生成时间: {current_time}\n内容: {content}",
    input_variables=["content"],
    # 将 "current_time" 变量与 get_current_datetime 函数进行绑定
    partial_variables={"current_time": get_current_datetime}
)

# 调用模板时，我们只需要提供 content 即可
current_time 会在 invoke 执行时，由 get_current_datetime 函数动态生成
final_prompt = prompt.invoke({"content": "一切正常。"})
print(final_prompt.to_string())

• 使用函数动态填充
  • 实现方法：预填充更强大的用法是使用函数动态地预填充变量，通过在模板构造函数中使用partial_variables参数绑定函数，可在模板调用时动态生成变量值。
  • 应用场景：适用于需要实时计算的变量，如当前时间、从数据库查询的信息等，通过动态填充可确保变量值的实时性和准确性。

from datetime import datetime
from langchain_core.prompts import PromptTemplate

# 定义一个函数，用于获取当前格式化的时间
def get_current_datetime():
    return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

# 在创建模板时，直接通过 partial_variables 参数传入函数本身
prompt = PromptTemplate(
    template="报告生成时间: {current_time}\n内容: {content}",
    input_variables=["content"],
    # 将 "current_time" 变量与 get_current_datetime 函数进行绑定
    partial_variables={"current_time": get_current_datetime}
)

# 调用模板时，我们只需要提供 content 即可
# current_time 会在 invoke 执行时，由 get_current_datetime 函数动态生成
final_prompt = prompt.invoke({"content": "一1111。"})
print(final_prompt.to_string())

5. 模版的持久化

• LangChain提供了将提示模板持久化到本地文件系统中的功能，支持JSON和YAML两种格式，便于团队协作、版本控制和跨项目复用。
• 持久化避免了在代码中硬编码大量提示字符串，使项目更加整洁和专业，提升了代码的可维护性和复用性。
• 应用场景：适用于精心设计并调试好的提示模板，通过持久化可方便地在不同项目或团队中共享和复用，提高开发效率。

from langchain_core.prompts import PromptTemplate, load_prompt

# 假设我们有一个想要保存的、效果很好的模板
my_template = PromptTemplate.from_template("为{product}写一句吸引人的广告语。")

# 将模板保存为 JSON 文件
# LangChain 会根据文件扩展名 .json 自动推断出正确的格式
my_template.save("my_advert_prompt.json")

# 在另一个地方，或者在未来的某个时间点，我们可以从文件轻松加载该模板
loaded_template = load_prompt("my_advert_prompt.json")

# 我们可以验证一下，加载的模板与原始模板是否完全相同
print(f"模板是否相同: {loaded_template == my_template}")

# 当然，加载后的模板可以像新创建的一样正常使用
print(loaded_template.invoke({"product": "自动咖啡机"}))

{
    "name": null,
    "input_variables": [
        "product"
    ],
    "optional_variables": [],
    "output_parser": null,
    "partial_variables": {},
    "metadata": null,
    "tags": null,
    "template": "\u4e3a{product}\u5199\u4e00\u53e5\u5438\u5f15\u4eba\u7684\u5e7f\u544a\u8bed\u3002",
    "template_format": "f-string",
    "validate_template": false,
    "_type": "prompt"
}

通过代码示例展示了如何将一个提示模板保存为JSON文件，并在其他地方或时间点从文件加载该模板，验证了加载后的模板与原始模板的相同性。