Langchain 学习笔记

官网：https://docs.langchain.com

初始化项目

uv init 

依赖安装

uv add langchain langchain-openai langchain-anthropic langchain-community

uv add langchain-redis redis

# 向量时候使用
uv add dashscope

# LangGraph
uv add langgraph

# 向量数据库
uv add chromadb
uv add langchain_ollama
uv add langchain_mongodb

加载.env

uv add python-dotenv

.env

APP_NAME=langchain
APP_URL=https://orangbus.cn

import os
from dotenv import load_dotenv

load_dotenv()
print(os.getenv("APP_NAME"))

依赖镜像

docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 -e REDIS_ARGS="--requirepass redis666" redis/redis-stack:latest

dockerfile

FROM python:3.14.2-slim

WORKDIR /app

# 复制依赖清单
COPY requirements.txt ./

# pip 安装依赖（--no-cache-dir 减少镜像体积）
RUN pip install --no-cache-dir -r requirements.txt

# 复制项目代码
COPY . .

# 暴露端口
EXPOSE 80

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

快速使用

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(
    model="Qwen/Qwen2.5-7B-Instruct",
    base_url="https://api.siliconflow.cn/v1",
    api_key="sk-otxxx"
)

基本使用

response = llm.invoke("你是谁？")
print(response.content)

消息

from langchain_core.messages import SystemMessage, HumanMessage

prompt1 = [
    SystemMessage("你的是回答问题的小助手"),
    HumanMessage("什么是着相？")
]
print(llm.invoke(prompt1))

prompt_template = ChatPromptTemplate.from_messages([
    {
            "role": "system",
            "content": [{"type": "text", "text": "你是一个考公的批阅老师，请根据用户上传的图片或者文字对学员的答案进行解析"}],
        },
        {
            "role": "user",
            "content": [
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "{img_url}"
                    },
                },
                {"type": "text", "text":'{content}'},
            ],
        },
])
parser = StrOutputParser()
chain = prompt_template | llm | parser
result = chain.invoke({
    "content":"请帮我批改一下图片中的申论作文 http://xx.com/images/2026/work.jpg"
})
print(result)

链式调用

def prompt():
    content = "html content"
    if content == None:
        return "获取内容失败"

    prompt_template = ChatPromptTemplate.from_messages([
        ("system", "你是一个html解析小助手， 你需要将用户输入的内容解析,严格按照json的格式返回，返回个字段如下：title:标题,content:提取文章的全部内容，desc: 正对正文的简介，20-30个字左右"),
        ("human", "{content}"),
    ])
    # print(prompt_template.invoke({"content": content}))
    
    parser = StrOutputParser()
    chain = prompt_template | llm | parser
    return chain.invoke({ "content": content})

print(prompt())

记忆对话

DB_URI = "postgres://postgres:postgres@homecc:5433/langchain_db?sslmode=disable&client_encoding=utf8"

with PostgresSaver.from_conn_string(DB_URI) as checkPointer:
    checkPointer.setup()
    
    

工具使用

获取外部数据等场景

from langchain.agents import create_agent
from langchain_core.tools import tool
from rich import print as rprint
import requests

@tool(parse_docstring=True)
def getMovieInfo(name: str):
    """根据电影的名称获取电影的信息,以及更新的集数

    Args:
        name: 电影名称

    Returns:
        json格式的电影的详细信息
    """
    response = requests.get(f"https://cj.rycjapi.com/api.php/provide/vod?ac=videolist&wd={name}")
    if response.status_code != 200:
        return f"获取电影信息错误，请求状态码是:{response.status_code}"
    data = response.json()
    if len(data["list"]) == 0:
        return "未查询到电影信息"
    movie =data["list"][0]
    del movie["vod_play_url"]
    return movie

from typing import Literal
from pydantic import BaseModel,Field
from langchain_core.utils.function_calling import convert_to_openai_function

class MovieParam(BaseModel):
    name: str = Field(description="电影的名称",default="仙逆")
    ac:str = Literal["video","videolist"] # 在列表中选择

# @tool(parse_docstring=True,args_schema=MovieParam)
def getMovieInfo2(name: str,ac:str="video"):
    """根据电影的名称获取电影的信息,以及更新的集数
    """
    response = requests.get(f"https://cj.rycjapi.com/api.php/provide/vod?ac={ac}&wd={name}")
    if response.status_code != 200:
        return f"获取电影信息错误，请求状态码是:{response.status_code}"
    data = response.json()
    if len(data["list"]) == 0:
        return "未查询到电影信息"
    movie =data["list"][0]
    del movie["vod_play_url"]
    return movie

rprint(convert_to_openai_function(getMovieInfo2))

调用工具

from app.llm.openai import llm
agent = create_agent(
    model=llm,
    tools=[getMovieInfo],
    system_prompt="帮助用户查询最新的电影信息"
)
result = agent.invoke({
    "messages":[
        {"role":"user","content":"斗罗大陆更新到第几集了？"}
    ]
})
rprint(result)

mcp

跟外界的系统打通场景

中间件

Model calllimit: 限制模型调用次数，防止一次任务反复请求LLM，导致费用失控 Tool call limit: 限制工具调用次数，避免Agent无限试错、死循环调工具

Summarization: 在上下文快满时自动总结历史，减少token消耗 Contextediting: 裁剪上下文、清理工具调用痕迹，本质上也是为了节省上下文成本

记忆存储

postgress

postgresql://name:passwd@127.0.0.1:5432/db_name?sslmodel=disable

import os, dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver
from rich import print as rprint

dotenv.load_dotenv()
user_id = "user_1"
db_url = "postgresql://postgres:admin666@home.cc:5433/langchain_db?sslmode=disable"

# # 构建模型
llm = ChatOpenAI(
    base_url=os.getenv("ZHIPU_URL"),
    model_name=os.getenv("ZHIPU_MODEL"),
    api_key=os.getenv("ZHIPU_API_KEY"),
)

with PostgresSaver.from_conn_string(db_url) as checkpointer:
    checkpointer.setup() # 初始化数据库

    agent = create_agent(
        model=llm,
        tools=[],
        checkpointer=checkpointer,
        system_prompt="当用户给你一个应用名称的时候，你返回对应的docker-compose生产环境的配置文件，并说明详细的配置"
    )
    result = agent.invoke({
        "messages": [{"role": "user", "content": "nginx"}],
    },
        config={"configurable": {"thread_id": user_id}}
    )
    rprint(result)

解析结构化数据

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field
from app.utils.utils import getUrlContent
import os

load_dotenv(override=True)

model = init_chat_model(
    model=os.getenv('MOTA_MODEL'),
    model_provider="openai",
    api_key=os.getenv("MOTA_API_KEY"),
    base_url=os.getenv("MOTA_URL"),
)
# print( model.invoke("你好"))

# 定义输出的结构
class Article(BaseModel):
    """文章信息"""
    title: str = Field(description="文章标题")
    content:str = Field(description="文章正文内容")
    push_at:str = Field(description="发布时间")
    files:list[dict[str,str]]  = Field(description="附件链接地址,附件名称，完整的链接地址")
content = getUrlContent("https://sft.yn.gov.cn/tzgg/400059.jhtml",'div','article-list')
# print(content)

# 关联结构
llm = model.with_structured_output(Article)
result = llm.invoke(f"请将以下内容转换为结构化数据，去除所有html标签，正文中不需要包含标题信息，不要修改正文的文字内容：{content}")

# 打印输出
print(type(result))
print(result.title)
print(result.push_at)
print(result.content)

文档加载器

from langchain_community.document_loaders import TextLoader,CSVLoader,JSONLoader

loader = TextLoader("docs/data.md")
docs = loader.load()
print(docs)

json加载

pdf加载

pip install pypdf

from langchain_community.document_loaders   import PyPDFLoader
from rich import print as rprint
loader = PyPDFLoader("docs/data.pdf",extraction_mode="plain")
docs = loader.load()
rprint(docs)

数据向量化

Agent

流式输出

完成一个步骤显示一段内容，不是逐字输出，比如说，调用了2个工具，每个工具完成了才会输出

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

model = init_chat_model(
    model=os.getenv('MODEL'),
    api_key=os.getenv("API_KEY"),
    base_url=os.getenv("BASE_URL"),
    model_provider="openai",
)
agent = create_agent(
    model=model,
    tools=[],
)
stream = agent.stream({
    "messages": [
        {"role": "user", "content": "给我一个nginx的线上环境的docker-compose文件"}
    ]
},stream_mode='updates')
for chunk in stream:
    print(chunk)