LLM API (补充)

非流式请求

Python

from dotenv import load_dotenv
from openai import OpenAI
import os

# 获取 env 环境变量
load_dotenv()
api_key = os.getenv("API_KEY")
base_url = os.getenv("BASE_URL")

client = OpenAI(
    api_key=api_key,
    base_url=base_url,
)

res = client.responses.create(
    model="deepseek-flash",
    input=[
        {
            # 角色定义
            "role": "system",
            "content": "用中文回答"
        },
        {
            "role": "user",
            "content": "who are you?"
        },
    ]
)

print(res.output_text)

流式请求和推理模型

Python

res = client.responses.create(
    model="deepseek-flash",
    input=[
        {
            # 角色定义
            "role": "system",
            "content": "用中文回答，每次回答不超10个汉字"
        },
        {
            "role": "user",
            "content": "who are you?"
        },
    ],
    stream=True,  # 流式返回
    reasoning={
        "effort": "low",    # 推理强度
    }
)

for event in res:
    if event.type == 'response.reasoning_text.delta':
        print(event.delta, end="")
    elif event.type == 'response.output_text.delta':
        print(event.delta, end="")

会话状态管理

OpenAI 官方 + 支持 Responses 的模型可以使用。DeepSeek 不支持，只有 Chat Completions，不存响应。

Python

if __name__ == '__main__':
    response_id = None
    while True:
        user_input = input()
        res = client.responses.create(
            model="deepseek-flash",
            input=user_input,
            previous_response_id=response_id
        )
        response_id = res.id
        print(f"response_id={response_id}")
        print(res.output_text)

JSON 输出

Python

response = client.responses.create(
    model="gpt-4o-mini",
    instructions="提取句子中的要素",
    input=[
        {
            "role": "user",
            "content": "1922年，尼克·卡拉威在纽约遇见了盖茨比。"
        }
    ],
    text={
        "format": {
            "type": "json_schema",
            "name": "timePlacePerson",
            "schema": {
                "type": "object",
                "properties": {
                    "time": { "type": "number" },
                    "place": { "type": "string" },
                    "person": { "type": "array", "items": { "type": "string" } }
                },
                "required": [ "time", "place", "person" ],
                "additionalProperties": False
            },
            "strict": True
        }
    }
)

print(response.output_text)

Tool Calls

Python

messages = []

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get weather of a location, the user should supply a location first. You need use English.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. Beijing",
                    }
                },
                "required": ["location"]
            },
        }
    }
]

def get_weather(location):
    if location == "Beijing":
        return "30"
    elif location == "Shanghai":
        return "28"
    else:
        return "0"

def send_msg():
    res = client.chat.completions.create(
        model="deepseek-flash",
        tools=tools,
        messages=messages
    )
    messages.append(res.choices[0].message)
    return res.choices[0].message


if __name__ == '__main__':
    user_input = "上海的温度是多少？"
    messages.append({"role": "user", "content": user_input})
    func_call_res = send_msg()
    tool = func_call_res.tool_calls[0]
    messages.append(
        {"role": "tool", "tool_call_id": tool.id, "content": get_weather(json.loads(tool.function.arguments)['location'])}
    )
    model_res = send_msg()
    print(model_res.content)

LangChain

LangChain 概述

LangChain 是一个 LLM 应用的开发框架，类似于 MyBatis，作为一中间层框架。

LangChain 架构：

LangChain
1. langchain-core：基础抽象类和 LangChain 表达式语言（LCEL）
2. langchain-community：社区维护的第三方集成
3. langchain：主包，包含链、代理、检索等高层应用编排逻辑
LangGraph：基于LangChain的进一步封装
LangSmith：可视化的监控，用于调试
LangServe：提供 restful 接口

RAG 检索增强生成：先从外部知识库中检索相关信息，再基于检索到的内容使用 LLM 生成答案

Anent = LLM + Memory + Tools + Planning + Action

LangChain 核心组件：

Model I/O：模型输入输出
Chains：链
Memory：记忆
Agents：智能体
Retrieval：检索
Callbacks：回调

Model I/O

OpenAI SDK

ChatCompletion API（常用）

Python

completion = client.chat.completions.create(
    model=model_name,
    messages=[{"role": "user", "content": "将'你好'翻译成意大利语"}],
)

print(completion.choices[0].message.content)

Responses API（新版）

Python

response  = client.responses.create(
    model=model_name,
    input="将'你好'翻译成意大利语"
)

print(response.output_text)

LangChain API

Python

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

load_dotenv()

llm = init_chat_model(
    model="deepseek-flash",
    model_provider="openai",
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY"),
)

# 传入字符串
res = llm.invoke("20字介绍若叶睦")
print(res.content)

# 传入消息列表
res = llm.invoke([
    SystemMessage("你的名字叫若叶睦"), HumanMessage("你叫什么名字")
])
print(res.content)

消息列表类别：SystemMessage、HumanMessage（用户）、AIMessage（AI）和 ToolMessage（Tools Calling）

消息列表还有列表写法

Python

[("system", "你的名字叫若叶睦"), ("user", "你叫什么名字")]

[{"role": "system", "content": "你的名字叫若叶睦"}, {"role": "user", "content": "你叫什么名字"}]

异步方法

Python

async def call():
    res = await llm.ainvoke("20字介绍若叶睦")
    print(res.content)

if __name__ == '__main__':
    asyncio.run(call())

流式返回

Python


res = llm.stream("300字介绍若叶睦")

for item in res:
    print(item.content, end="")

JSON 解析

使用 OpenAI SDK

Python

response = client.responses.create(
    model="gpt-4o-mini",
    instructions="提取句子中的要素",
    input=[
        {
            "role": "user",
            "content": "1922年，尼克·卡拉威在纽约遇见了盖茨比。"
        }
    ],
    text={
        "format": {
            "type": "json_schema",
            "name": "timePlacePerson",
            "schema": {
                "type": "object",
                "properties": {
                    "time": { "type": "number" },
                    "place": { "type": "string" },
                    "person": { "type": "array", "items": { "type": "string" } }
                },
                "required": [ "time", "place", "person" ],
                "additionalProperties": False
            },
            "strict": True
        }
    }
)

print(response.output_text)

使用LangChain 的 Prompt 约束

Python

class User(BaseModel):
    name: str = Field()
    desc: str = Field()
    friends: list[str] = Field()

user_parser = JsonOutputParser(pydantic_object=User)

res = llm.invoke([
    SystemMessage(user_parser.get_format_instructions()), HumanMessage("介绍一下若叶睦")
])
print(res.content)

使用LangChain 的大模型参数约束

Python

class User(BaseModel):
    name: str = Field()
    desc: str = Field()
    friends: list[str] = Field()

user_llm = llm.with_structured_output(schema=User)

user = user_llm.invoke("介绍一下若叶睦")
print(user)

提示词模板

Python

temp = ChatPromptTemplate.from_messages(
    messages=[
        ("system", "你是一个专业的动漫人物讲解员"),
        ("human", "请讲解{name}。"),
    ],
)
prompt = temp.invoke({"name": "若叶睦"})
res = llm.invoke(prompt)
print(res.content)

Chain

RunnableSequence 可运行序列

Python

llm = init_chat_model(
    model="deepseek-flash",
    model_provider="openai"
)
temp = ChatPromptTemplate.from_messages(
    messages=[
        ("system", "你是一个专业的动漫人物讲解员。50字介绍"),
        ("human", "请讲解{name}。"),
    ],
)
parser = StrOutputParser()

# 使用多个 Runnables 创建 RunnableSequence
agent = temp | llm | parser

res = agent.invoke({'name': '若叶睦'})
print(res)

RunnableParallel 可运行并行

Python

temp = ChatPromptTemplate.from_messages(
    messages=[
        ("system", "你是一个专业的动漫人物讲解员。50字介绍"),
        ("human", "请讲解{name}。"),
    ],
)

deepseek_llm = init_chat_model(
    model="deepseek-flash",
    model_provider="openai"
)

glm_llm = init_chat_model(
    model="glm-5.3-flash",
    model_provider="openai"
)

parser = StrOutputParser()

deepseek_chain = (temp | deepseek_llm | parser)
glm_chain = (temp | glm_llm | parser)

summary_chain = PromptTemplate.from_template("总结一下，输出1：{output1}，输出2：{output2}") | deepseek_llm | parser

map_chain = ({'output1': deepseek_chain, 'output2': glm_chain}) | summary_chain

res = map_chain.invoke({'name': '若叶睦'})
print(res)

Retrieval

RAG 介绍

RAG 检索增强生成

解决的问题：幻觉、领域知识、知识冻结

RAG 的流程：

知识库构建
1. 加载文档，非结构化数据解析
2. 文档切分
3. Embedding
4. 存到向量数据库
检索召回
1. query 的 embedding
2. 使用检索算法去数据库查询
3. chunks
4. rerank 重排序
5. 获取精排结构
生成 LLM：PE 提示词工程、SFT 监督微调、RLHF 强化学习

文档加载

使用 UnstructuredMarkdownLoader 加载 Markdown

Bash

pip install markdown langchain_community unstructured[md]

Python

from langchain_community.document_loaders import UnstructuredMarkdownLoader

# "elements"：按文档结构切分成多块
# "single"  ：整体作为一块，去掉 md 标记
loader = UnstructuredMarkdownLoader("./assert/deeplearning.md", mode="elements")
docs = loader.load()

使用 UnstructuredWordDocumentLoader 加载 doc

Bash

pip install unstructured[docx]

Python

from langchain_community.document_loaders import UnstructuredWordDocumentLoader

loader = UnstructuredWordDocumentLoader("./assert/langchain.docx", mode="elements")
doc = loader.load()

使用 Mineru 解析 PDF，可以使用 VLM 模型进行文档解析，解析之后可以得到 md。

https://mineru.net/

文档切分

将完整的文档进行分块处理（Chunking），以这些块（Chunk）为基本单位存储和检索

切分策略：

固定字符
递归使用多个分隔符切分，能保证不切断完整的句子
语义切分

RecursiveCharacterTextSplitter 递归字符文本切分器：按段落→句子→词逐级递归切分文本，尽量保持句子完整，并可用重叠来维持语义连贯

Bash

pip install langchain-text-splitters

Python

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = UnstructuredMarkdownLoader("./assert/deeplearning.md", mode="single")

docs = loader.load()

chunks = RecursiveCharacterTextSplitter(
    chunk_size=400, chunk_overlap=50, length_function=len
).split_documents(docs)

文档嵌入

Sentence-BERT 模型：把文本转换向量，从而快速计算句子之间的语义相似度。

Python

from langchain_community.embeddings import OpenAIEmbeddings

embed_model = OpenAIEmbeddings(
    base_url="https://api.siliconflow.cn/v1",
    api_key="sk-xxx",
    model="Qwen/Qwen3-Embedding-8B"
)

text = "你好，我是若叶睦"
res = embed_model.embed_query(text)
print(len(res)) # 4096
print(res)

text_list = ['你好我是若叶睦', "你好，我是丰川祥子"]
res_list = embed_model.embed_documents(text_list)
print(len(res_list))    # 2
print(len(res_list[0])) # 4096

稠密向量和稀疏向量区别：

稠密向量：负责兜底理解。用户说的词和原文对不上也没关系，它能靠语义把“意思差不多”的都捞回来，防止漏掉。
稀疏向量：负责精准命中。用户要找特定关键词、型号、代号时，它逐词核对，确保“原词出现过”的才被算进来，防止跑偏。

使用 Embedding 转换向量

BAAI/bge-m3：https://www.modelscope.cn/models/BAAI/bge-m3

Bash

pip install FlagEmbedding==1.3.5

Python

from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel(model_name_or_path=r"D:\llm\bge-m3")

text = "你好，我是若叶睦"

res = model.encode(text, return_dense=True, return_sparse=True)

# 稀疏向量
print(res["lexical_weights"])

# 稀疏向量转换为 token
print(model.convert_id_to_token(res["lexical_weights"]))

# 稠密向量
print(res["dense_vecs"])

向量存储和检索

向量数据库：Milvus

Shell

# Docker(Win) 启动
curl --ssl-revoke-best-effort -o standalone.bat https://raw.githubusercontent.com/milvus-io/milvus/refs/heads/master/scripts/standalone_embed.bat
standalone.bat start

Milvus 创建 Collection（对应 MySQL 的 Table）

Python

from pymilvus import MilvusClient, DataType

client = MilvusClient(uri="http://localhost:19530")

collection_name = "collection_demo"

client.drop_collection(collection_name=collection_name)

print(client.list_collections())

schema = (MilvusClient.create_schema(auto_id=True)
          .add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
          .add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024)  # 稠密向量 res["dense_vecs"]
          .add_field(field_name="sparse_vector", datatype=DataType.SPARSE_FLOAT_VECTOR)  # 稀疏向量 res["lexical_weights"]
          .add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1024)
          .add_field(field_name="meta", datatype=DataType.JSON))

index = MilvusClient.prepare_index_params()
index.add_index(field_name="vector", index_type="HNSW", metric_type="L2")
index.add_index(field_name="sparse_vector", index_type="SPARSE_INVERTED_INDEX", metric_type="IP")

client.create_collection(
    collection_name=collection_name,
    schema=schema,
    index_params=index,
)

print(client.list_collections())

Milvus 插入

Python

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pymilvus import MilvusClient, DataType

# 兼容性修复：transformers 5.x 移除了 is_torch_fx_available，
# 而 FlagEmbedding 1.3.5 仍会导入它，导致 ImportError。
import transformers.utils.import_utils as _iu
if not hasattr(_iu, "is_torch_fx_available"):
    def _is_torch_fx_available():
        try:
            import torch.fx  # noqa: F401
            return True
        except Exception:
            return False
    _iu.is_torch_fx_available = _is_torch_fx_available

from FlagEmbedding import BGEM3FlagModel

client = MilvusClient(uri="http://localhost:19530")

# print(client.list_collections())    # ['collection_demo']

collection_name = "collection_demo"

'''
schema = (MilvusClient.create_schema(auto_id=True)
          .add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
          .add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024)  # 稠密向量 res["dense_vecs"]
          .add_field(field_name="sparse_vector", datatype=DataType.SPARSE_FLOAT_VECTOR)  # 稀疏向量 res["lexical_weights"]
          .add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1024)
          .add_field(field_name="meta", datatype=DataType.JSON))
'''

loader = UnstructuredMarkdownLoader("./assert/deeplearning.md", mode="single")
md_data = loader.load()
if len(md_data) < 1:
    raise RuntimeError

text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50, separators=["\n\n", "\n", "。"])
md_split = text_splitter.split_text(md_data[0].page_content)

model = BGEM3FlagModel(model_name_or_path=r"D:\llm\bge-m3")

encode_list = list()
for split in md_split:
    res = model.encode(split, return_dense=True, return_sparse=True)
    encode_list.append({
        "vector": res["dense_vecs"],
        "sparse_vector": res["lexical_weights"],
        "text": split,
        "meta": None
    })
    
insert_res = client.insert(collection_name, encode_list)
print(insert_res)

Milvus 删除

Python

client.delete(
    collection_name=collection_name,
    # 过滤条件
    filter="id in [463480757150366907, 463480757150366908]",
)

Milvus 向量检索

Python

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pymilvus import MilvusClient, DataType, AnnSearchRequest, RRFRanker

# 兼容性修复：transformers 5.x 移除了 is_torch_fx_available，
# 而 FlagEmbedding 1.3.5 仍会导入它，导致 ImportError。
import transformers.utils.import_utils as _iu
if not hasattr(_iu, "is_torch_fx_available"):
    def _is_torch_fx_available():
        try:
            import torch.fx  # noqa: F401
            return True
        except Exception:
            return False
    _iu.is_torch_fx_available = _is_torch_fx_available

from FlagEmbedding import BGEM3FlagModel

client = MilvusClient(uri="http://localhost:19530")

COLLECTION_NAME = "collection_demo"

'''
schema = (MilvusClient.create_schema(auto_id=True)
          .add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
          .add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024)  # 稠密向量 res["dense_vecs"]
          .add_field(field_name="sparse_vector", datatype=DataType.SPARSE_FLOAT_VECTOR)  # 稀疏向量 res["lexical_weights"]
          .add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1024)
          .add_field(field_name="meta", datatype=DataType.JSON))

index = MilvusClient.prepare_index_params()
index.add_index(field_name="vector", index_type="HNSW", metric_type="L2")
index.add_index(field_name="sparse_vector", index_type="SPARSE_INVERTED_INDEX", metric_type="IP")
'''

model = BGEM3FlagModel(model_name_or_path=r"D:\llm\bge-m3")


def query_pymilvus(query: str, limit: int = 5):
    query_encode = model.encode(query, return_dense=True, return_sparse=True)
    dense_vec = query_encode["dense_vecs"]  # 稠密向量
    sparse_vec = query_encode["lexical_weights"]  # 稀疏向量
    # BGEM3 返回的稀疏向量 key 是字符串形式的 token id，Milvus 要求 int 下标
    # sparse_vec = {int(k): v for k, v in sparse_vec.items()}

    # 稠密向量搜索
    # index.add_index(field_name="vector", index_type="HNSW", metric_type="L2")
    res_dense_vecs = client.search(
        collection_name=COLLECTION_NAME,
        data=[dense_vec],
        anns_field="vector",
        limit=limit,
        search_params={"metric_type": "L2"},
        output_fields=["id", "text"]
    )
    print("稠密向量搜索结果: ", res_dense_vecs)
    print("=" * 50)

    # 稀疏向量搜索
    # index.add_index(field_name="sparse_vector", index_type="SPARSE_INVERTED_INDEX", metric_type="IP")
    res_sparse_vector = client.search(
        collection_name=COLLECTION_NAME,
        data=[sparse_vec],
        anns_field="sparse_vector",
        limit=limit,
        search_params={"metric_type": "IP"},
        output_fields=["id", "text"]
    )
    print("稀疏向量搜索结果: ", res_sparse_vector)
    print("=" * 50)

    # 混合向量搜索
    dense_req = AnnSearchRequest(
        data=[dense_vec],
        anns_field="vector",
        param={"metric_type": "L2"},
        limit=limit,
    )
    sparse_req = AnnSearchRequest(
        data=[sparse_vec],
        anns_field="sparse_vector",
        param={"metric_type": "IP"},
        limit=limit,
    )

    results = client.hybrid_search(
        collection_name=COLLECTION_NAME,
        reqs=[dense_req, sparse_req],
        ranker=RRFRanker(k=60),
        limit=limit,
        output_fields=["id", "text"]
    )
    print("混合向量搜索结果: ", results)
    return results



if __name__ == '__main__':
    query = "权重共享"
    result = query_pymilvus(query)
    print(result)

Milvus 标量检索

Python

client.query(
    collection_name=collection_name,
    filter=f'text like "%{keyword}%"',
    output_fields=["id", "text"],
    limit=5,
)

生成

利用 PE 提示词工程，插入查询到的信息。

Python

message_list = [
    {
        "role": "system",
        "content": "你是一个专业的法律问答机器人，请根据上下文回答问题，当上下文无法回答问题时，请回答“根据上下文无法回答该问题”"
    },
    {
        "role": "user",
        "content": f"根据以下上下文回答问题：{context}\n问题：{query}"
    }
]

Agent

Agent 介绍

Anent = LLM + Memory + Tools + Planning + Action

本地工具

使用 OpenAI SDK (Tool Calls)

Python

messages = []

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get weather of a location, the user should supply a location first. You need use English.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. Beijing",
                    }
                },
                "required": ["location"]
            },
        }
    }
]

def get_weather(location):
    if location == "Beijing":
        return "30"
    elif location == "Shanghai":
        return "28"
    else:
        return "0"

def send_msg():
    res = client.chat.completions.create(
        model="deepseek-flash",
        tools=tools,
        messages=messages
    )
    messages.append(res.choices[0].message)
    return res.choices[0].message


if __name__ == '__main__':
    user_input = "上海的温度是多少？"
    messages.append({"role": "user", "content": user_input})
    func_call_res = send_msg()
    tool = func_call_res.tool_calls[0]
    messages.append(
        {"role": "tool", "tool_call_id": tool.id, "content": get_weather(json.loads(tool.function.arguments)['location'])}
    )
    model_res = send_msg()
    print(model_res.content)

使用 langchain 定义工具

Python

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

# 获取 env 环境变量
load_dotenv()


# 这里必须要有 docstring
@tool
def get_weather(location) -> str:
    """获取指定城市的天气温度。

    Args:
        location: 城市名称，例如 "Beijing"。
    """
    if location == "Beijing":
        return "30"
    else:
        return "0"


llm = init_chat_model(
    model="deepseek-flash",
    model_provider="openai",
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY"),
)

llm = llm.bind_tools([get_weather])

message_list = [
    HumanMessage(content="北京的温度是多少")
]

res = llm.invoke(message_list)
message_list.append(res)     # 这里需要把 AIMessage 加进去

tool_call = res.tool_calls[0]
args = tool_call['args']    # args: {'location': 'Beijing'}
tool_id = tool_call['id']   # tool_id: call_00_nWl4A52AXMZo6MdVHZQk3942

weather = get_weather.invoke(args)    # weather: 30

message_list.append(ToolMessage(tool_call_id=tool_id, content=weather))
res = llm.invoke(message_list)  # res.content: 北京当前的温度是 30°C。

MCP

MCP：模型上下文协议，大语言模型与数据的通信标准

MCP 分类：stdio（本地执行）和 http（服务器执行）

MCP 工作流程：

同步工具列表
上下文注入发给 LLM
LLM 决策来选择工具调用
工具执行

stdio 形式的 MPC

Python

# server
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Weather")

@mcp.tool()
def get_weather(city: str) -> str:
    return f"{city}是18摄氏度"

if __name__ == '__main__':
    mcp.run(transport="stdio")

Python

# client
import asyncio

from mcp import StdioServerParameters, stdio_client, ClientSession

params = StdioServerParameters(
    command=r"C:\ProgramData\anaconda3\envs\langchain\python.exe",
    args=[r"./17_mcp_stdio_server.py"]
)


async def mcp_get_weather(city: str):
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            return await session.call_tool("get_weather", {"city": city})

if __name__ == '__main__':
    weather = asyncio.run(mcp_get_weather("北京"))
    print(weather)

http 形式的 MPC

Python

# server
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Weather")

@mcp.tool()
def get_weather(city: str) -> str:
    return f"{city}是18摄氏度"

if __name__ == '__main__':
    mcp.run(transport="streamable-http")  # port: 8000

Python

# client
import asyncio

from mcp import StdioServerParameters, stdio_client, ClientSession
from mcp.client.streamable_http import streamable_http_client

mcp_url = r"http://localhost:8000/mcp"

async def mcp_get_weather(city: str):
    async with streamable_http_client(url=mcp_url) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()

            return await session.call_tool("get_weather", {"city": city})

if __name__ == '__main__':
    weather = asyncio.run(mcp_get_weather("北京"))
    print(weather)

LangChain 的 MPC

Python

import asyncio
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient

load_dotenv()

mcp = MultiServerMCPClient({
    "get_weather": {
        "transport": "streamable_http",
        "url": "http://localhost:8000/mcp"
    }
})


async def main(query: str):
    tools = await mcp.get_tools()
    llm = init_chat_model(
        model="deepseek-flash",
        model_provider="openai",
        base_url=os.getenv("OPENAI_BASE_URL"),
        api_key=os.getenv("OPENAI_API_KEY"),
    )
    agent = create_agent(model=llm, tools=tools)
    res = await agent.ainvoke({"messages": [HumanMessage(query)]})
    print(res)

if __name__ == '__main__':
    asyncio.run(main("北京"))

记忆

指定一个 thread_id，即可将短期记忆存储到内存中的消息列表。每个 thread_id 的会话历史相互隔离。

Python

import asyncio
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

llm = init_chat_model(
    model="deepseek-flash",
    model_provider="openai",
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY"),
)

checkpointer = InMemorySaver()
thread_id = "114514"


async def call():
    agent = create_agent(model=llm, checkpointer=checkpointer)
    res = await agent.ainvoke({"messages": [HumanMessage("20字介绍一下若叶睦")]},
                              config={"configurable": {"thread_id": thread_id}})
    print("第一次回答：", res)
    res = await agent.ainvoke({"messages": [HumanMessage("她所在乐队的键盘手是谁？")]},
                              config={"configurable": {"thread_id": thread_id}})
    print("第二次回答：", res)


if __name__ == '__main__':
    asyncio.run(call())

中间件

上下文压缩：SummarizationMiddleware

Python

import asyncio
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

llm = init_chat_model(
    model="deepseek-flash",
    model_provider="openai",
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY"),
)

checkpointer = InMemorySaver()
thread_id = "114514"

summary_middleware = SummarizationMiddleware(
    model=llm,
    trigger=("messages", 3),    # 触发值
    keep=("messages", 2),       # 保留值
)


async def call():
    agent = create_agent(model=llm, checkpointer=checkpointer, middleware=[summary_middleware])
    while True:
        query = input()
        res = await agent.ainvoke({"messages": [HumanMessage(query)]},
                                  config={"configurable": {"thread_id": thread_id}})
        print(res['messages'][-1].content)
        print("len(res['messages']):", len(res['messages']))


if __name__ == '__main__':
    asyncio.run(call())

工具调用人工确认：HumanInTheLoopMiddleware

Python

LangChain：LLM 应用开发框架https://wmhwiki.cn/posts/2026/langchain

作者wmh

发布于9/28

更新于今日内

许可协议 CC BY-NC-SA 4.0

署名-非商业性使用-相同方式共享 4.0 国际

AI技术 LangChain

LangChain：LLM 应用开发框架

LLM API (补充) ​

非流式请求 ​

流式请求和推理模型 ​

会话状态管理 ​

JSON 输出 ​

Tool Calls ​

LangChain ​

LangChain 概述 ​

Model I/O ​

OpenAI SDK ​

LangChain API ​

JSON 解析 ​

提示词模板 ​

Chain ​

Retrieval ​

RAG 介绍 ​

文档加载 ​

文档切分 ​

文档嵌入 ​

向量存储和检索 ​

生成 ​

Agent ​

Agent 介绍 ​

本地工具 ​

MCP ​

记忆 ​

中间件 ​

LLM API (补充)

非流式请求

流式请求和推理模型

会话状态管理

JSON 输出

Tool Calls

LangChain

LangChain 概述

Model I/O

OpenAI SDK

LangChain API

JSON 解析

提示词模板

Chain

Retrieval

RAG 介绍

文档加载

文档切分

文档嵌入

向量存储和检索

生成

Agent

Agent 介绍

本地工具

MCP

记忆

中间件