LangChain由Harrison Chase创建于2022年10月,是用于开发智能体工程的平台。
1、Agent:在人工智能领域,Agent(智能体或代理)是指一种能感知环境、进行推理、自主决策并采取行动以实现特定目标的智能系统。也就是给大模型装上大脑
2、Tool(工具):底层可调用的函数 / Api,粒度最小;比如查时间、调用计算器、调用搜索API;输入输出结构化
3、Skill(技能):多个Tool组合封装出来的某件任务的能力。例如“查询并总结网页资料”这个Skill,由【网页搜索Tool + 网页解析Tool】组合而成
4、Action(动作):Agent 向外执行的一步操作。可以是调用 Tool,也可以输出自然语言回复。Tool 调用属于 Action 的其中一类。
5、Observation(观测):Tool执行之后返回给Agent的结果。Agent拿到Observation,继续思考
6、tool_call(工具调用):大模型输出的一段结构化JSON指令;用来告诉Agent,我们要调用什么工具,什么指令
Tought -> Action -> Observation->Thought
一、LangChain调用大模型的大概流程
1、安装LangChain依赖
uv add langchain # 有pyproject.toml文件下用 #正常情况下我们还是用pip pip install langchain如果没有uv环境还得先装uv环境
pip install uv2、安装各种不同的模型,每个模型提供对应的兼容SDK
#集成Deepseek uv add langchain-deepseek pip install langchain-deepseek #集成Open AI uv add langchain-openai pip install langchain-openai #集成 Anthropic uv add langchain-anthropic pip install langchain-anthropic3、加载环境变量(这个在部署和调用大模型的那个文章中讲过就是解析.env文件的)
from dotenv import load_dotenv load_dotenv()4、定义工具(Tool)
from langchain.tools import tool @tool def getWeather(location:str) -> str: """ 得到天气情况的工具 tool中必须有这个解释 """ return f"Current weather in {location} is sunny"5、创建Agent
from langchain.agents import creat_agent agent = creat_agent( "deepseek-chat", tools = [getWeather] )6、调用Agent
response = agent.invoke({ "messages":[ {"role":"user","content":"今天杭州天气如何?"} ] }) for message in respone['messages']: print(message.model_dump_json(indent=2)) #model_dump_json是转换成json格式,indent=2是遇到逗号换行,遇到下一层嵌套在换行前空2格7、Agent系统的执行流程
用户提问(input)-> 模型分析(Reasoning)-> 调用工具(Action)->感知分析(Observation)
-> 生成结果(Output)
总的来说,大模型用来接收消息思考并且输出tool_call,决定用哪个tool,把这些信息组装成tool_calls返回,然后LangChain根据分析tool_call去Action执行Tool,再把Observation的结果交给大模型,让它再次思考输出结果。
二、模型(Models)
1、初始化模型(langchain支持的模型)
(1)langchain支持市面上大部分的大语言模型,并提供了各个模型对应的依赖库。
OpenAI(gpt)-> langchain-openai -> ChatOpenAI、
Anthropic(claude)-> langchain-anthropic -> ChatAnthropic、
Google(gemini)-> langchain-google-genai -> ChatGoogleGenerativeAI
DeepSeek(deepseek)-> langchain-deepseek -> ChatDeepSeek
from langchain_openai import ChatOpenAI model = ChatOpenAI(model = "gpt-5.2")ps:这种方式是老方式,因为你要用几个大模型就得导入几个类,再去创建几个对象,不适配于多模型切换的场景。
(2)为解决上面的弊端引入了“init_chat_model”这个函数
这个函数不需要导入任何函数,只需要指定你想要的模型即可。
from dotenv import load_dotenv from langchain.chat_models import init_chat_model load_dotenv ('config.env') llm = init_chat_model( model='deepseek-flash' temperature = 1.5 ) print(type(llm))2、如果langchain里没有对应的库
(1)需要自定义模型参数来访问,如:qwen是openai旗下的
from dotenv import load_dotenv from langchain.chat_models import init_chat_model import os load_dotenv ('config.env') api_key = os.getenv("DASHSCOPE_API_KEY") base_url = os.getenv("DASHSCOPE_BASE_URL") llm = init_chat_model( model='qwen-max', model_provider="openai" # langchain不认识的模型就需要指明提供商 api_key = api_key, base_url = base_url temperature = 1.5 )(2)用老方法 -- langchain社区中的库
pip install langchain-community uv add langchain-communitypip install dashscope #阿里云白炼自己的库from langchain_community.chat_models.tongyi import ChatTongyi model = ChatTongyi( model = "qwen-max" )3、调用模型
因为langchain可以在初始化的时候把模型参数初始化固定(temparature等),所以我们调用模型的时候就只需要传递消息即可
(1)invoke完整内容输出到显示器
from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent import os load_dotenv ('config.env') llm = init_chat_model( model='deepseek-flash', temperature = 1.5 ) response = llm.invoke([ { "role":"system", "content":"你扮演火箭队的武藏" }, { "role":"user", "content":"你是谁" } ]) 如果想采用流式调用,直接把invoke改成stream即可:llm.stream print(response.content)如果直接:response = llm.invoke("你是谁?"),默认就是user的内容
(2)流式,一个字一个字的输出
from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent import os load_dotenv ('config.env') llm = init_chat_model( model='deepseek-flash', temperature = 1.5 ) response = llm.stream([ { "role":"system", "content":"你扮演火箭队的武藏" }, { "role":"user", "content":"你是谁" } ]) 如果想采用流式调用,直接把invoke改成stream即可:llm.stream for l in response: print(l.content,end="",flush=True) # end=“”遍历一次不换行,flush刷新缓冲机制,True可以让它输出内容后马上显示4、在智能体中调用模型
(1)创建智能体:creat_agent
可以先初始化再在智能体里用
res = create_agent( model=llm )也可以直接
res = create_agent( model="deepseek-falsh" )(2)用智能体调用模型
invoke型:
from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent import os load_dotenv ('config.env') llm = init_chat_model( model='deepseek-flash', temperature = 1.5 ) agent = create_agent(model=llm) response = agent.invoke({ "messages":[{"role":"user","content":"你是谁"}] }) print(response["messages"][-1].content)stream型:
from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent import os load_dotenv ('config.env') llm = init_chat_model( model='deepseek-flash', temperature = 1.5 ) agent = create_agent(model=llm) response = agent.stream({ "messages":[{"role":"user","content":"你是谁"}] }, stream_mode="messages" #stream有很多种类型,要想看到回复的消息流式输出,就是messages模式 ) for token,metadata in response: if token.content: print(token.content,end="",flush=True)三、消息(Message)
在langchain中,发送给模型的消息、模型返回的消息都统一被封装成BaseMessage,并准备了多个BaseMessage的子类对应不用角色类型的消息。
1、BaseMessage的子类
(1)SystemMessage -> Role:system
(2)HumanMessage -> Role:user
(3)AIMessage -> Role:assistant(模型返回的消息)
(4)ToolMessage -> Role:tool
response = agent.invoke({ "messages":[ {"role":"system","content":"你是虎哥"}, {"role":"user","content":"你是谁"}, {"role":"assistant","content":"你好呀,我是虎哥"}, {"role":"user","content":"今天杭州的天气怎么样"} ] })就可以直接用BaseMessage
from langchain_core.messages import SystemMessage,HumanMessage,AIMessage response = agent.invoke({ "messages":[ SystemMessage(content = "你是虎哥"), HumanMessage(content = "你是谁"), AIMessage(content = "你好呀,我是虎哥"), HumanMessage(content="今天杭州的天气怎么样") ] })总代码就是:
from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain.tools import tool from langchain_core.messages import SystemMessage,HumanMessage,AIMessage load_dotenv ('config.env') 初始化模型 llm = init_chat_model( model='deepseek-flash', temperature = 1.5 ) @tool def getWeather(location:str)->str: """summary Returns: str: 返回天气 """ return f"今天{location}的天气是晴朗" #创建一个Langchain对象 agent = create_agent(model=llm,tools=[getWeather]) #用Langchain创建出的对象去调用大模型 response = agent.invoke({ "messages":[ SystemMessage(content = "你是虎哥"), HumanMessage(content = "你是谁"), AIMessage(content = "你好呀,我是虎哥"), HumanMessage(content="今天杭州的天气怎么样") ] }) for message in response['messages']: message.pretty_print() #更好看的打印信息用pretty_print()方法打印的就是这样
2、多模态消息
很多大模型不止能处理文字消息,还能处理图片等多模态的信息,但是前提是这个模型本身就是多模态模型,deepseek就不行,但是上面提到的qwen就可以识别图片。
(1)在线图片
from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain.tools import tool from langchain_core.messages import SystemMessage,HumanMessage,AIMessage import os load_dotenv ('config.env') llm = init_chat_model( model="qwen-vl-max", api_key = os.getenv('DASHSCOPE_API_KEY'), base_url="https://maas.qianwenaiapi.com/compatible-mode/v1", model_provider='openai', temperature = 1.5 ) agent = create_agent(model=llm) response = agent.stream({ "messages":[ HumanMessage(content=[ {"type": "image_url","image_url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"}, {"type": "text","text": "图中描绘的是什么景象?"} ]) ] }, stream_mode='messages' ) for chunk,meta in response: if chunk.content: print(chunk.content,end='',flush=True)(2)本地图片
有时用户会上传磁盘的图片数据,而不是url地址,我们就需要把图片数据转换成base64字符串,然后发送给模型。
首先安装一个上传组件,用于模拟图片上传(上传必须有前端界面或者用python的.ipynb文件):
!uv add ipywidgetsfrom ipywidgets import FileUpload from IPython.display import display Which file types to accept, e.g. '.doc,.docx'.*代表所有类型 multiple是否允许上传多个文件,这里我们演示就先设置成False upload = FileUpload(accept = '*',mutiple = False) display(upload) ''' ({'name': 'OIP.webp', 'type': 'image/webp', 'size': 12898, 'content': <memory at 0x0000018A5E374DC0>, 'last_modified': datetime.datetime(2026, 10, 9, 9, 31, 46, 203000, tzinfo=datetime.timezone.utc)}, ) '''ps:其中upload.value的content就是存放base64编码的地址
上传后,就要把图片数据转换成base64的数据格式:
import base64 # 拿取第一张照片的内容的值 upload_file = upload.value[0] # 把存放内容的内存地址赋值给content_dizhi content_dizhi = upload_file['content'] # 将内存地址的内容转化为bytes类型 image_bytes = bytes(content_dizhi) # 将bytes类型的内容转化为base64编码 image_base64 = base64.b64encode(image_bytes).decode('utf-8')再编写消息message并调用大模型
from langchain_core.messages import HumanMessage message = HumanMessage(content=[ {"type": "text", "text": "请帮我分析这张图片的内容"}, { "type": "image", "base64": image_base64, "mime_type": "image/jpeg", }, ]) response = agent.stream({ "messages": [message] }, stream_mode="messages" ) for chunk,data in response: if chunk.content: print(chunk.content,end="",flush=True)四、提示词(prompts)
提示词(prompts)就是发送给模型的消息,其中SystemMessage是系统提示词(system prompt),可以给模型设定角色、聊天背景、任务说明,对模型生成的内容有很大的影响。
1、在message中设置SystemMessage
同样的问题,系统提示词不一样回答的问题也不一样,比如:
message = [SystemMessage(content="你以海盗的口吻回答问题"), HumanMessage(content="你是谁?")]回答:嘿嘿!俺是纵横七海、闻着宝藏味儿就能找到路的海盗船长!帆一扯、炮一响,连海鸥都得给俺让道。你问俺是谁?先报上名来,免得俺的弯刀认错了人
message = [SystemMessage(content="你以台湾女子的口吻回答问题"), HumanMessage(content="你是谁?")]回答:我是林小满啦,台湾女生,今年28岁,住台中,平时喜欢喝手摇饮、逛夜市,也爱跟朋友聊聊天。你这样问我,是想认识我一下吗?😊
2、在agent中设置全局系统提示词
这样就不用每次传消息都要设置系统提示词了
agent = create_agent( model=llm, system_prompt="以海盗的口吻回答我的问题。" ) message = [HumanMessage(content="你是谁?")]3、提示词工程
通过优化提示词,使模型输出的结果更符合业务的逻辑。
身份角色(Identity):描述AI的职责、沟通风格和总体目标
指令说明(Instructions):指导模型如何生成所需的响应,它应该遵循哪些规则?模型应该做什么,以及模型绝对不能做什么
对话示例(Examples):提供可能的输入示例,以及模型期望的输出
背景信息(Context):向模型提供生成响应所需的任何额外信息,例如RAG的额外知识库数据,或你认为特别相关的任何其他数据
编写System Prompt时,推荐使用Markdown格式和XML标签的组合来帮助模型理解提示和上下文数据逻辑的边界(避免话太多,逻辑太杂乱,风格不喜欢)
(1)身份角色和指令说明
agent = create_agent( model=llm, system_prompt=""" #身份:你是一个编程助手,你帮用户编写python文件 #指令:你只需要输出python代码,不要输出任何解释和多余的文字 """ ) message = [HumanMessage(content="怎样定义string变量记录人民,比如:张飞")](2)对话示例
agent = create_agent( model=llm, system_prompt=""" #身份:你是一个科幻作家,根据用户的要求创建一个太空之都 #示例: user:月球的首都是什么 assistant:月球的首都是“月光城”,位于月球的南极地区,是一个高科技、环保、宜居的城市。 user:火星的首都是什么 assistant:火星的首都是“火星城”,位于火星的赤道地区,是一个充满未来感和科技感的城市。 """ ) message = [HumanMessage(content="木星的首都是什么")]回答:木星的首都是“云穹城”,位于木星赤道云带的高层大气中,是一座悬浮在气态巨行星上空的浮空都市,依靠反重力平台和风暴能源运转,充满未来科技感。
(3)结构化输出
a、基于提示词的结构化输出
agent = create_agent( model=llm, system_prompt=""" #身份:你是一个科幻作家,根据用户的要求创建一个太空之都 #指令:务必以json格式输出,不要加任何markdown语法和多余的文字,json中必须包含城市名称、位置、特点三个字段 #示例: user:月球的首都是什么 assistant: {"name":"月华市" "location":"位于月球正面赤道附近的静海面基地遗址上" "vibe":"冷调、高效、革新" "economy":"量子通信枢纽、尖端生物圈" } """ ) message = [HumanMessage(content="木星的首都是什么")]回答:
{
"name":"云冠市",
"location":"位于木星赤道北带高空云层中的反重力悬浮都市群,依托轨道环木卫三中转港连接",
"vibe":"恢宏、炽烈、精密",
"economy":"氦-3采集与磁层能源、星际航运枢纽、气态巨行星深。空科研"
}
b、基于Model的结构化输出
·定义一个数据类型(基于pydantic)
·创建智能体,设置输出格式
from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from dotenv import load_dotenv from pydantic import BaseModel import os load_dotenv('Api.env') llm = init_chat_model( model="deepseek-chat", model_provider="openai", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com", temperature = 0.7 ) # 定义一个继承BaseModel的类,写属性 class City(BaseModel): name: str location: str vibe: str economy: str agent = create_agent( model=llm, system_prompt="你是一个科幻作家,根据用户的要求创建一个太空之都", response_format=City #设置结构化的输出格式,pydantic校验大模型输出必须是JSON格式,且有City中定义的属性,否则就报错 ) message = [HumanMessage(content="木星的首都是什么")] response = agent.invoke({"messages": message} ) print(response['structured_response'])五、工具(Tools)
一个完整的Agent至少包含2个关键的部分:大模型(推理分析、规划任务),工具(执行任务)
定义工具 -> 初始化模型 -> 初始化Agent,绑定模型和工具
1、自定义工具
本质就是一个可调用的函数用@tool装饰的,自定义要包括:工具名、工具作用、工具需要的参数
from langchain.tools import tool @tool(args_schema= WeatherInput) def get_weather(city: str,units:str = "celsius") -> str: """ Description: Get the current weather for a given city. Args: city (str): The name of the city to get the weather for. units (str): The units for the temperature (e.g., "celsius", "fahrenheit"). """ return f"{city}天气29{units},适合出行。"可以通过Pydantic Model描述参数,去约束参数,在tool工具前加上pydantic描述类:
from pydantic import BaseModel,Field from typing import Literal #枚举 class WeatherInput(BaseModel): city: str = Field(..., description="The name of the city to get the weather for.") units: Literal["celsius", "fahrenheit"] = Field( default="celsius", description="温度单位,摄氏度或华氏度")2、预定义工具
很多常用的工具其实已经被写好了,langchain官网的文档中就有很多预定义的tool
我来用其中的Tavily Search这个搜索工具演示一下怎么调用,这个工具的作用是,能够快速提供实时、准确且事实相关的结果。(让智能体联网)
step1:点进去看一下这个tool的文档步骤,按步骤先下载它的SDK包
step2:这个工具里面有很多功能,用最常见的搜索功能举例可以搞一个tool函数,把代码封装进去,也可以用langchain官方的方法
(1)tool函数封装
from langchain.tools import tool from tavily import TavilyClient import os @tool def search(message: str): """ A simple search tool that returns the input message. """ tavily_client = TavilyClient(api_key=os.getenv("tvly-YOUR_API_KEY")) response = tavily_client.search(message) return reponse(2)用langchain封装好的方法
首先还是下载相关SDK导入key,但是方法和上面的不一样,具体在langchain官网查看
from dotenv import load_dotenv load_dotenv('Api.env') import getpass import os if not os.environ.get("TAVILY_API_KEY"): os.environ["TAVILY_API_KEY"] = getpass.getpass("Tavily API key:\n")再用它官方封装的tool
from langchain_tavily import TavilySearch #相关参数可以在文档里查看 tool = TavilySearch( max_results=5, topic="general", # include_answer=False, # include_raw_content=False, # include_images=False, # include_image_descriptions=False, # search_depth="basic", # time_range="day", # include_domains=None, # exclude_domains=None )最后用智能体调用
from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain_core.messages import HumanMessage import os llm = init_chat_model( model="deepseek-chat", model_provider="openai", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com" ) agent = create_agent( model=llm, tools=[tool] ) response = agent.invoke({"messages": [HumanMessage(content="Who is Leo Messi,用中文回答")]}) print(response["messages"][-1].content)(3)优化
a、token消耗大
Langchain官方提供的这个tool包含了完整的参数列表,会导致额外的Token消耗,因此,对于简单的业务,可以用tavily封装一下。
from langchain_tavily import TavilySearch from langchain.tools import tool #相关参数可以在文档里查看 tavily_searcher= TavilySearch( max_results=5, topic="general" ) @tool def search(message: str): """ Args: message (str): The input message to be searched. """ res=tavily_searcher.invoke({"query": message}) return res["results"]这样的话,在大模型调用tool时,tool已经调用invoke了tavily搜索工具返回的是字典中的results,把其他的东西全部清洗掉了,再传输给大模型思考,就会节省token。如果让智能体干完全程返回的就是一大段字典,如下图
b、结果中不包含网页数据源,可行度低,容易出现幻觉胡写
可以强制输出时结构化输出,用上面所说的pydantic的BaseModel类
from langchain_tavily import TavilySearch from langchain.tools import tool from pydantic import BaseModel, Field #相关参数可以在文档里查看 tavily_searcher= TavilySearch( max_results=5, topic="general" ) #Agent回答内容引用的网页信息 class RefreshInput(BaseModel): title:str=Field(...,description="The title of the document to be refreshed.") url:str=Field(...,description="The URL of the document to be refreshed.") #Agent的回答内容 class SearchInput(BaseModel): answer:str=Field(...,description="The answer to the user's question.") references:list[RefreshInput]=Field(...,description="The list of references used to answer the user's question.") def search(message: str): """ Args: message (str): The input message to be searched. """ res=tavily_searcher.invoke({"query": message}) return res["results"]就会输出图示的信息,把每个信息的来源的url网页地址都标注清楚,可信度就很高了