参考资料
- 快速入门视频 以及该教程对应的“笔记文章” juejin.cn/post/752699…
- langChain官方文档
- langGragh基于langChain的高级封装框架，快速搭建agents
- 给langGraph搭建的agents可视化展示
- langSmith，用于agents监控的框架
- langFlow，对标dify的AI工作流产品，相当于langChain带可视化界面
核心功能
- 是什么？
- LangChain 是一个开发大预言模型的框架。它可以将 LLM 与外部数据源、工具和各种组件便捷地连接起来，构建功能强大且实用的应用。
- 特点
- LangChain目前主要的功能
-
模型接口 封装OpenAI、Claude、Cohere、Qwen、deepseek等模型统一调用方式
-
输出结构化 自动从模型中解析JSON、Schema、函数签名、文档等
-
内存****管理 Buffer, Summary, Entity, Conversation Memory
-
工具接入 Web 搜索、SQL数据库、Python 执行器、API代理等
-
Agent 架构 ReAct、Self-Ask、OpenAl Function Agent 等调度机制
-
RAG 集成 多种 Retriever、Vector Store、文档拆分策略
- 支持的语言
- LangChain主要支持python语言，部分支持js语言，因此我们可以在我们的前端全栈项目中引入LangChain开发智能体
- 应用-可以用于开发
接入模型
LangChain实现了标准化接口，实现了很多AI厂商的接入sdk，这里以deepseek为例，使用LangChain接入并调用一个简单工具
注意，LangChain不同版本差异较大，本文使用的版本是0.3.25
不同的模型使用不同的接入sdk即可，代码中就可以统一使用init_chat_model初始化模型
安装langchain和langchain-deepseek
pip install langchain langchain-deepseek
接入API并调用
from langchain.chat_models import init_chat_model
model = init_chat_model(
model="deepseek-chat",
model_provider="deepseek",
base_url="https://api.deepseek.com",
api_key="sk-xxx"
)
res = model.invoke('你好')
print(res.content)
链式调用
我们将大模型、相关工具等作为组件，链就是负责将这些组件按照某一种逻辑，顺序组合成一个流水线的方式
模版提示词
langchain内置了多种提示词组件，可以让AI按要求回答问题，比如常见的提示词组件有
- ChatPromptTemplate
- PromptTemplate
from ast import parse
from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
model = init_chat_model(
model="deepseek-chat",
model_provider="deepseek",
base_url="https://api.deepseek.com",
api_key="sk-xxx"
)
prompt_template = ChatPromptTemplate([
("system", "你是一个乐意助人的助手，请根据用户的问题给出回答"),
("user", "这是用户的问题： {topic}， 请用 yes 或 no 来回答")
])
# 直接使用模型 + 输出解析器
bool_qa_chain = prompt_template | model | StrOutputParser()
# 测试
question = "请问 1 + 1 是否 大于 2？"
result = bool_qa_chain.invoke({'topic':question})
print(result)
上述简单代码，即可结构化输出LLM的结果
暂时无法在飞书文档外展示此内容
结构化解析
提到结构化解析，langChain内置的解析器包括
| 解析器名称 |
功能描述 |
类型 |
| BooleanOutputParser |
将LLM输出解析为布尔值 |
基础类型解析 |
| DatetimeOutputParser |
将LLM输出解析为日期时间 |
基础类型解析 |
| EnumOutputParser |
解析输出为预定义枚举值之一 |
基础类型解析 |
| RegexParser |
使用正则表达式解析LLM输出 |
模式匹配解析 |
| RegexDictParser |
使用正则表达式将输出解析为字典 |
模式匹配解析 |
| StructuredOutputParser |
将LLM输出解析为结构化格式 |
结构化解析 |
| YamlOutputParser |
使用Pydantic模型解析YAML输出 |
结构化解析 |
| PandasDataFrameOutputParser |
使用Pandas DataFrame格式解析输出 |
数据处理解析 |
| CombiningOutputParser |
将多个输出解析器组合为一个 |
组合解析器 |
| OutputFixingParser |
包装解析器并尝试修复解析错误 |
错误处理解析 |
| RetryOutputParser |
包装解析器并尝试修复解析错误 |
错误处理解析 |
| RetryWithErrorOutputParser |
包装解析器并尝试修复解析错误 |
错误处理解析 |
| ResponseSchema |
结构化输出解析器的响应模式 |
辅助类 |
比如使用BooleanOutputParser解析器，将上面的示例代码，解析为布尔值
from langchain.output_parsers import BooleanOutputParser
# 直接使用模型 + 输出解析器
bool_qa_chain = prompt_template | model | BooleanOutputParser()
# 测试
question = "请问 1 + 1 是否 大于 2？"
result = bool_qa_chain.invoke({'topic':question})
print(result)
使用更频繁的解析器有StructedOutputParser，他可以让AI从文本中提取结构化数据
from ast import parse
from langchain.chat_models import init_chat_model
from langchain.prompts import PromptTemplate
from langchain.output_parsers import StructuredOutputParser, ResponseSchema
model = init_chat_model(
model="deepseek-chat",
model_provider="deepseek",
base_url="https://api.deepseek.com",
api_key="sk-xxx"
)
schemas = [ # 构建结构化数据模板
ResponseSchema(name="name", description="用户的姓名"),
ResponseSchema(name="age", description="用户的年龄")
]
parser = StructuredOutputParser.from_response_schemas(schemas)
prompt = PromptTemplate(
template="请根据以下内容提取用户信息，并返回 JSON 格式：\n{input}\n\n{output}"
)
chain = (
prompt.partial(output=parser.get_format_instructions())
| model
| parser
)
result = chain.invoke({"input": "用户叫李雷，今年25岁，是一名工程师。"})
print(result) # {'name': '李雷', 'age': '25'}
多条链组合
langChain还可以定义多个chain，形成工作流形式，处理数据
from ast import parse
from langchain.chat_models import init_chat_model
from langchain.prompts import PromptTemplate
from langchain.output_parsers import StructuredOutputParser, ResponseSchema
model = init_chat_model(
model="deepseek-chat",
model_provider="deepseek",
base_url="https://api.deepseek.com",
api_key="sk-xxx"
)
news_gen_prompt = PromptTemplate.from_template(
"请根据以下标题撰写一段文章内容，包括时间地点事件具体信息：\n\n标题：{title}"
)
# 第一个子链：生成新闻内容
news_chain = news_gen_prompt | model
schemas = [
ResponseSchema(name="time", description="事件发生的时间格式化为YYYY-MM-DD"),
ResponseSchema(name="location", description="事件发生的地点"),
ResponseSchema(name="event", description="发生的具体事件"),
]
parser = StructuredOutputParser.from_response_schemas(schemas)
summary_prompt = PromptTemplate.from_template(
"请从下面这段文章内容中提取关键信息，并返回结构化JSON格式：\n\n{news}\n\n{output}"
)
# 第二个子链：生成新闻摘要
summary_chain = (
summary_prompt.partial(output=parser.get_format_instructions())
| model
| parser
)
# 组合成一个复合 Chain
full_chain = news_chain | summary_chain
# 调用复合链
result = full_chain.invoke({"title": "小米公司在武汉总部发布小米16Pro，搭载骁龙8gen5"})
print(result)
自定义组件
如果我们需要在两个chain中间进行调试，就可以使用自定义组件
from langchain_core.runnables import RunnableLambda
def debug_print(x):
print('中间结果（新闻正文）：', x)
return x
debug_node = RunnableLambda(debug_print)
# 组合成一个复合 Chain
full_chain = news_chain | debug_node | summary_chain
记忆功能
使用 MessagesPlaceholder 在 ChatPromptTemplate 中为“对话历史”留出一个占位符，把用户/助手的历史消息按顺序插入到系统消息之后，从而构建完整的聊天消息列表给模型
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
model = init_chat_model(
model="deepseek-chat",
model_provider="deepseek",
base_url="https://api.deepseek.com",
api_key="sk-xxx"
)
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个专业的前端开发工程师，你喜欢玩弄前端八股，时常会说出一些让人难以理解的技术名词"),
MessagesPlaceholder(variable_name="messages"),
])
chain = prompt | model | StrOutputParser()
messages_list = [] # 初始化历史
while True:
user_query = input("你：")
# 1) 追加用户消息
messages_list.append(HumanMessage(content=user_query))
# 2) 调用模型
assistant_reply=''
print('AI:', end=' ')
for chunk in chain.stream({"messages": messages_list}):
assistant_reply+=chunk
print(chunk, end="", flush=True)
print()
# 3) 追加 AI 回复
messages_list.append(AIMessage(content=assistant_reply))
调用工具
langChain官方内置了很多工具，可以参考文档
python.langchain.com/docs/integr…
使用AI调用工具是一个相对比较麻烦的过程，首先需要给AI绑定工具，让AI知道有哪些工具可以调用
tools = [get_weather]
llm_with_tools = model.bind_tools(tools)
执行AI对话后，AI会返回内容，其中包含需要调用的工具
{
"content": "我来帮您查询南昌今天的天气情况。",
... 省略其他内容
"tool_calls": [
{
"name": "get_weather",
"args": {
"loc": "南昌"
},
"id": "call_0_15fd694f-a1d1-4187-b325-792794ca71ac",
"type": "tool_call"
}
],
... 省略其他内容
}
这个时候在代码中要判断AI返回结果是否有工具调用请求，如果有则需要找到对应的工具函数，调用，并在再次调用AI，并填入函数调用结果。
暂时无法在飞书文档外展示此内容
而langChain中的agents功能简化了AI调用工具的流程，他可以自动执行调用功能步骤，并自动调用AI回答
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
import requests
import json
model = init_chat_model(
model="deepseek-chat",
model_provider="deepseek",
base_url="https://api.deepseek.com",
api_key="sk-xxx"
)
@tool
def get_weather(loc):
"""
查询即时天气函数
:param loc: 必要参数，字符串类型，用于表示查询天气的具体城市名称，\
:return：心知天气 API查询即时天气的结果，具体URL请求地址为："https://api.seniverse.com/v3/weather/now.json"
返回结果对象类型为解析之后的JSON格式对象，并用字符串形式进行表示，其中包含了全部重要的天气信息
"""
url = "https://api.seniverse.com/v3/weather/now.json"
params = {
"key": "xxx-xxx",
"location": loc,
"language": "zh-Hans",
"unit": "c",
}
response = requests.get(url, params=params)
temperature = response.json()
return temperature['results'][0]['now']
tools = [get_weather]
prompt = ChatPromptTemplate.from_messages(
[
("system", "你是天气助手，请根据用户的问题，给出相应的天气信息"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"), # 这部分agent提示符写法是写死的不可以修改
]
)
agent = create_tool_calling_agent(llm=model, tools=tools, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
response = agent_executor.invoke({"input": "南昌今天的天气如何？"})
print(json.dumps(response, indent=2, ensure_ascii=False))
Agent
上一节中，就通过Agent调用了我们自定义的工具
LangChain Agent API 在底层对 大模型调用工具做了优化，让我们在代码中也可以方便的使用AI并集成工具
除了create_tool_calling_agent这一通用的方法，LangChain还封装了许多不同的Agent实现形式，参考下表:
| 函数名 |
功能描述 |
适用场景 |
| create_tool_calling_agent |
创建使用工具的Agent |
通用工具调用 |
| create_openai_tools_agent |
创建OpenAI工具Agent |
OpenAI模型专用 |
| create_openai_functions_agent |
创建OpenAI函数Agent |
OpenAI函数调用 |
| create_react_agent |
创建ReAct推理Agent |
推理+行动模式 |
| create_structured_chat_agent |
创建结构化聊天Agent |
多输入工具支持 |
| create_conversational_retrieval_agent |
创建对话检索Agent |
检索增强对话 |
| create_json_chat_agent |
创建JSON聊天Agent |
JSON格式交互 |
| create_xml_agent |
创建XML格式Agent |
XML逻辑格式 |
| create_self_ask_with_search_agent |
创建自问自答搜索Agent |
自主搜索推理 |
MCP调用
MCP（全称是Model Context Protocol，模型上下文协议）
MCP统一了AI调用工具的格式，也让AI能更好的拓展工具，常规的Function Calling工具是写在源码中不可变动的
通过MCP可以自由拓展AI的能力
在langChain中使用mcp，需要先安装依赖
pip install langchain-mcp-adapters
再调用高德地图mcp工具
import asyncio
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_mcp_adapters.client import MultiServerMCPClient
import json
mcp_config = json.load(open('./mcp.json', 'r', encoding='utf-8')).get('mcpServers', {})
mcp_client = MultiServerMCPClient(mcp_config)
model = init_chat_model(
model="deepseek-chat",
model_provider="deepseek",
base_url="https://api.deepseek.com",
api_key="sk-xxx",
)
prompt = ChatPromptTemplate.from_messages(
[
("system", "你是一个地图工具，能够提供实时的地图信息和导航服务。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"), # 这部分agent提示符写法是写死的不可以修改
]
)
async def run():
tools = await mcp_client.get_tools()
agent = create_tool_calling_agent(llm=model, tools=tools, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
response = await agent_executor.ainvoke({"input": "南昌的天气怎么样"})
print(json.dumps(response, indent=2, ensure_ascii=False))
asyncio.run(run())
登录
