基于langchain源码剖析常见用法

1
2
3
4
5
6
7
8
9
from langchain_classic.agents import initialize_agent, AgentType
from langchain_openai import OpenAI
from langchain_core.tools import Tool

llm = OpenAI(temperature=0)
tools = [Tool(name="Calculator", func=lambda x: eval(x), description="用于计算数学问题")]
agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

1
2
3
4
5
from langchain_classic.agents import create_react_agent
from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(...)  # 自定义 ReAct 提示词
agent = create_react_agent(llm, tools, prompt)

1
2
result = agent.invoke({"input": "3的平方加上5的立方等于多少？"})
print(result["output"])  # 输出计算结果

1
2
3
4
5
6
7
8
9
10
11
12
# ReAct模式，多步骤推理
AgentType.ZERO_SHOT_REACT_DESCRIPTION
# 基于zere-shot reAct，引入memory，适合聊天型工具调用
AgentType.CONVERSATIONAL_REACT_DESCRIPTION
# 工具描述比ReAct更严格，输出强制JSON
AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION
# OpenAI Function Calling流行后退出的agent，适合生产环境
AgentType.OPENAI_FUNCTIONS
# 长任务拆分agent，适合大任务，多步骤自动化执行
from langchain.agents import PlanAndExecute
# 存在历史基于，有向量库，主要用于知识库，企业文档AI助手之类的RAG系统
ConversationalRetrievalChain  

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

model = init_chat_model(
    "claude-sonnet-4-5-20250929",
    # Kwargs passed to the model:
    temperature=0.7,
    timeout=30,
    max_tokens=1000,
)

model = ChatOpenAI(
    model="gpt-5",
    temperature=0.1,
    max_tokens=1000,
    timeout=30
    # ... (other params)
)

# 创建智能体
graph = create_agent(
    model="anthropic:claude-sonnet-4-5-20250929",
    tools=[check_weather],
    system_prompt="You are a helpful assistant",
)

# 调用智能体
inputs = {"messages": [{"role": "user", "content": "what is the weather in sf"}]}
for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@wrap_model_call
def fallback_model(request, handler):
    try:
        return handler(request)  # 尝试主模型
    except Exception:
        # 切换到备用模型
        request = request.override(model=fallback_model_instance)
        return handler(request)

agent = create_agent(
    model=basic_model,  # Default model
    tools=tools,
    middleware=[fallback_model]
)

1
2
3
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
response = llm.invoke("介绍一下 LangChain")

1
2
3
4
5
6
7
8
9
10
from langchain.tools import Tool

def get_user(user_id: str):
    return {"user": user_id, "name": "Zhou Jiang"}

user_tool = Tool(
    name="get_user_info",
    func=get_user,
    description="Get user info by user_id"
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langchain.tools import StructuredTool
from pydantic import BaseModel

class UserInput(BaseModel):
    user_id: int
    verbose: bool

def fetch_user(user_id: int, verbose: bool):
    return {"id": user_id, "detail": verbose}

user_tool = StructuredTool.from_function(
    func=fetch_user,
    args_schema=UserInput,
    name="fetch_user",
)

1
2
3
# 基础文本生成
prompt = PromptTemplate.from_template("Write a {length} poem about {topic}.")
prompt_text = prompt.format(length="short", topic="spring")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate

system_template = "You are a helpful assistant that translates {input_language} to {output_language}."
human_template = "Translate: {text}"

# AIMessagePromptTemplate：AI 回复（用于示例）（一般很少使用）
system_prompt = SystemMessagePromptTemplate.from_template(system_template)
human_prompt = HumanMessagePromptTemplate.from_template(human_template)
chat_prompt = ChatPromptTemplate.from_messages([system_prompt, human_prompt])

result = chat_prompt.format_prompt(
    history="", # 历史消息
    input_language="English",
    output_language="French",
    text="Hello world"
).to_messages()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain.prompts import PromptTemplate
from langchain.prompts.example_selector import NGramOverlapExampleSelector

examples = [
    {"query": "如何学习Python编程", "response": "建议从基础语法开始..."},
    {"query": "Python入门教程", "response": "推荐使用官方文档..."},
    {"query": "Java学习方法", "response": "先掌握面向对象概念..."}
]

example_prompt = PromptTemplate(
    input_variables=["query", "response"],
    template="Query: {query}\nResponse: {response}"
)

# 初始化N-gram选择器（2-gram，选择1个最佳示例）
selector = NGramOverlapExampleSelector(
    examples=examples,
    example_prompt=example_prompt,
    k=1,
    ngram_size=2
)

# 输入：与“Python学习”相关的查询
input = {"query": "Python学习路径"}
selected_example = selector.select_examples(input)
print(selected_example[0]["query"])  # 输出："如何学习Python编程"（与输入共享更多2-gram）

1
2
3
from langchain_classic.chains import LLMChain
chain = LLMChain(prompt=prompt, llm=llm)
result = chain.invoke({"question": "LangChain 有哪些特点？"})

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
prompt1 = ChatPromptTemplate.from_messages([
    ("system", "将中文翻译成英文。"),
    ("human", "{chinese_text}")
])
chain1 = LLMChain(llm=ChatOpenAI(), prompt=prompt1, output_key="english_text")

prompt2 = ChatPromptTemplate.from_messages([
    ("system", "将英文翻译成法语。"),
    ("human", "{english_text}")
])
chain2 = LLMChain(llm=ChatOpenAI(), prompt=prompt2, output_key="french_text")

sequential_chain = SequentialChain(
    chains=[chain1, chain2],
    input_variables=["chinese_text"],
    output_variables=["french_text"]
)
result = sequential_chain.run(chinese_text="我喜欢编程")

1
2
3
4
5
class ConversationChain(LLMChain):
    memory: BaseMemory = Field(default_factory=ConversationBufferMemory)
    prompt: BasePromptTemplate = PROMPT
    input_key: str = "input"  # 用户输入的键名
    output_key: str = "response"  # 模型输出的键名

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 初始化向量数据库（示例数据）
documents = [
    Document(page_content="LangChain 支持多种链类型，包括 LLMChain、SequentialChain 等。"),
    Document(page_content="RetrievalQA 链用于检索增强问答，结合检索器和语言模型。")
]
embeddings = OpenAIEmbeddings()
db = FAISS.from_documents(documents, embeddings)
retriever = db.as_retriever()

# 创建 RetrievalQA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(),
    chain_type="stuff",  # 使用 Stuff 策略合并文档
    retriever=retriever,
    return_source_documents=True  # 返回用于回答的源文档
)

# 执行问答
result = qa_chain({"query": "LangChain 有哪些链类型？"})
print(result["result"])  # 输出：LangChain 支持多种链类型，包括 LLMChain、SequentialChain 等。
print("源文档：", [doc.page_content for doc in result["source_documents"]])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# 定义两个子链和路由规则
math_prompt = ChatPromptTemplate.from_messages([
    ("system", "解决数学问题：{input}")
])
code_prompt = ChatPromptTemplate.from_messages([
    ("system", "生成 Python 代码：{input}")
])
router_template = """
根据输入判断类型：
- 若涉及数学计算，返回 `math`
- 若涉及代码生成，返回 `code`
输入：{input}
"""
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    output_parser=RouterOutputParser()
)

# 初始化路由链和子链
llm = ChatOpenAI()
router_chain = LLMRouterChain.from_llm(llm, router_prompt)
math_chain = LLMChain(llm=llm, prompt=math_prompt)
code_chain = LLMChain(llm=llm, prompt=code_prompt)
chain_map = {"math": math_chain, "code": code_chain}
multi_prompt_chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=chain_map,
    default_chain=LLMChain(llm=llm, prompt=ChatPromptTemplate.from_messages([("human", "{input}")]))
)

1
2
3
4
5
6
7
8
# 每轮：
recent = window_memory.format()
retrieved = retriever.get_relevant_docs(query, k=4)
prompt = system + recent + format_retrieved(retrieved) + user_question
answer = llm(prompt)
# 保存新消息到 memory 和向量库（upsert）
memory.save(user_message, assistant_message)
vectorstore.add_texts([user_message, assistant_message], metadata=...)

1
2
3
4
5
6
7
8
9
10
from langchain_classic.memory import CombinedMemory, SimpleMemory

# 组合对话记忆和简单键值对记忆
conv_memory = ConversationBufferMemory(memory_key="history")
simple_memory = SimpleMemory(memories={"user_name": "小明"})
combined = CombinedMemory(memories=[conv_memory, simple_memory])

combined.save_context({"input": "你好"}, {"output": "您好！"})
print(combined.load_memory_variables({}))
# 输出：{"history": "Human: 你好\nAI: 您好！", "user_name": "小明"}

1
2
3
4
from langchain.embeddings import OpenAIEmbeddings
from langchain.embeddings import AzureOpenAIEmbeddings
from langchain.embeddings import GPT4AllEmbeddings
from langchain.embeddings import HuggingFaceEmbeddings

1
2
3
4
5
6
7
8
9
10
from langchain_community.embeddings import HuggingFaceHubEmbeddings
from langchain_community.embeddings import SentenceTransformerEmbeddings
from langchain_community.embeddings import CohereEmbeddings
from langchain_community.embeddings import GooglePalmEmbeddings
from langchain_community.embeddings import BedrockEmbeddings
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.embeddings import MistralAIEmbeddings
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.embeddings import VoyageEmbeddings
from langchain_community.embeddings import JinaEmbeddings

1
2
3
4
5
6
embedding = embeddings.embed_query("hello world")
vectors = embeddings.embed_documents(["a", "b", "c"])

# langchain新标准
embeddings.invoke("text")
embeddings.batch(["t1", "t2"])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# 文档加载，加载csv格式的数据
from langchain.document_loaders import CSVLoader
file = 'OutdoorClothingCatalog_1000.csv'
loader = CSVLoader(file_path=file)
docs = loader.load()

# OpenAIEmbeddings对文本进行向量化
# 调用 OpenAI 的嵌入 API（默认使用text-embedding-ada-002模型），将文本转换为 1536 维向量。
from langchain.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()

# 创建向量索引
# 将向量库转换为检索器，提供统一的检索接口，用于根据用户查询获取相关文档。
from langchain.vectorstores import DocArrayInMemorySearch
db = DocArrayInMemorySearch.from_documents(
    docs, 
    embeddings
)

# 将索引转换为检索器，其实是将索引作为检索器的一个变量。检索器提供了不同相关性计算的方法
retriever = db.as_retriever()

# 创建检索式问答链
from langchain.chat_models import ChatOpenAI
llm_model = 'gpt-3.5-turbo-0301' # 后续该模型会下线，替换成其他模型即可
llm = ChatOpenAI(temperature = 0.0, model=llm_model)

from langchain.chains import RetrievalQA
qa_stuff = RetrievalQA.from_chain_type(
    llm=llm, 
    chain_type="stuff", 
    retriever=retriever, 
    verbose=True
)
qa_stuff.query("List all your shirts with sun protection in a table")

1
2
3
4
5
6
7
8
9
10
11
12
13
from langchain_core.callbacks import StdOutCallbackHandler
from langchain_openai import ChatOpenAI

handler = StdOutCallbackHandler()
llm = ChatOpenAI(
    model="deepseek-chat",  
    temperature=0,
    callbacks=[handler],  # 直接传入 Handler 列表（自动创建 CallbackManager）
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL
)

llm.invoke("Hello World")

1
2
3
4
5
6
from langchain_core.callbacks import FileCallbackHandler
import sys

# 写入到标准输出（等同于 StdOutCallbackHandler）
# 或写入到文件：handler = FileCallbackHandler("logs.txt")
handler = FileCallbackHandler(sys.stdout)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain_core.callbacks import StreamingStdOutCallbackHandler
from langchain_openai import ChatOpenAI

# 配置 DeepSeek 模型参数
DEEPSEEK_API_KEY = "sk-65da967e427c4f86ae4749129ba48166"  # 替换为你的 DeepSeek API Key
DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1"  # DeepSeek 的 API 基础地址

# 初始化 DeepSeek 模型，启用流式输出
llm = ChatOpenAI(
    model="deepseek-chat",  # 指定 DeepSeek 模型，可选 deepseek-chat/deepseek-coder
    temperature=0.7,  # 控制生成随机性，0 为确定性输出，1 为最大随机性
    streaming=True,  # 必须开启，否则无法流式输出
    callbacks=[StreamingStdOutCallbackHandler()],  # 绑定流式回调处理器，控制台会实时打印内容
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL
)

# 执行生成任务，控制台会实时打印内容
llm.invoke("介绍你的功能")