LangChain 底层原理#

1
Prompt → Model → OutputParser
2
  ↓        ↓         ↓
3
构造提示词  调用LLM   解析输出

1
# 伪代码：Runnable 中 | 的实现原理
2
class Runnable:
3
    def __or__(self, other):
4
        return RunnableSequence(self, other)
5

6
    def __ror__(self, other):
7
        return RunnableSequence(coerce_to_runnable(other), self)

1
# pip install langchain langchain-openai
2

3
from langchain_core.prompts import ChatPromptTemplate
4
from langchain_core.output_parsers import StrOutputParser
5
from langchain_openai import ChatOpenAI
6

7
# 1. 定义提示模板
8
prompt = ChatPromptTemplate.from_messages([
9
    ("system", "你是一位专业的技术写作助手。"),
10
    ("human", "请用一段话解释什么是 {concept}。")
11
])
12

13
# 2. 定义模型
14
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
15

16
# 3. 定义输出解析器
17
parser = StrOutputParser()
18

19
# 4. 用 LCEL 管道串联
20
chain = prompt | model | parser
21

22
# 5. 调用
23
result = chain.invoke({"concept": "向量数据库"})
24
print(result)

1
# pip install langchain langchain-openai
2

3
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
4
from langchain_core.prompts import ChatPromptTemplate
5
from langchain_core.output_parsers import StrOutputParser
6
from langchain_openai import ChatOpenAI
7

8
prompt = ChatPromptTemplate.from_template(
9
    "根据以下上下文回答问题。\n上下文: {context}\n问题: {question}"
10
)
11
model = ChatOpenAI(model="gpt-4o-mini")
12

13
# 模拟检索函数
14
def fake_retriever(input_dict):
15
    return "LangChain 是一个用于构建 LLM 应用的框架。"
16

17
# RunnablePassthrough.assign() 可以在透传的同时附加新字段
18
chain = (
19
    RunnablePassthrough.assign(context=lambda x: fake_retriever(x))
20
    | prompt
21
    | model
22
    | StrOutputParser()
23
)
24

25
result = chain.invoke({"question": "什么是 LangChain？"})
26
print(result)

1
# pip install langchain langchain-openai
2

3
from langchain_core.runnables import RunnableParallel
4
from langchain_core.prompts import ChatPromptTemplate
5
from langchain_core.output_parsers import StrOutputParser
6
from langchain_openai import ChatOpenAI
7

8
model = ChatOpenAI(model="gpt-4o-mini")
9

10
# 两个不同的处理分支
11
summary_chain = (
12
    ChatPromptTemplate.from_template("用一句话总结: {text}")
13
    | model
14
    | StrOutputParser()
15
)
16

17
keywords_chain = (
18
    ChatPromptTemplate.from_template("提取3个关键词（逗号分隔）: {text}")
19
    | model
20
    | StrOutputParser()
21
)
22

23
# 并行执行两个分支
24
parallel_chain = RunnableParallel(
25
    summary=summary_chain,
26
    keywords=keywords_chain
27
)
28

29
result = parallel_chain.invoke({
30
    "text": "LangChain 是一个用于构建大语言模型应用的开源框架，"
31
            "提供了模块化的组件和丰富的集成。"
32
})
33

34
print(result)
35
# 输出: {"summary": "...", "keywords": "..."}

1
# pip install langchain
2

3
from langchain_core.runnables import RunnableLambda
4

5
# 普通 Python 函数
6
def word_count(text: str) -> dict:
7
    words = text.split()
8
    return {"text": text, "word_count": len(words)}
9

10
# 包装为 Runnable
11
word_counter = RunnableLambda(word_count)
12

13
# 支持完整的 Runnable 协议
14
result = word_counter.invoke("LangChain 是一个强大的框架")
15
print(result)  # {"text": "LangChain 是一个强大的框架", "word_count": 5}
16

17
# 也支持批量调用
18
results = word_counter.batch(["你好世界", "Hello World LangChain"])
19
print(results)

1
# pip install langchain langchain-openai
2

3
from langchain_core.runnables import RunnableBranch, RunnableLambda
4
from langchain_core.prompts import ChatPromptTemplate
5
from langchain_core.output_parsers import StrOutputParser
6
from langchain_openai import ChatOpenAI
7

8
model = ChatOpenAI(model="gpt-4o-mini")
9

10
# 定义不同话题的处理链
11
tech_chain = (
12
    ChatPromptTemplate.from_template("你是技术专家。请回答: {question}")
13
    | model | StrOutputParser()
14
)
15

16
general_chain = (
17
    ChatPromptTemplate.from_template("你是通用助手。请回答: {question}")
18
    | model | StrOutputParser()
19
)
20

21
# 话题分类函数
22
def is_tech_question(input_dict: dict) -> bool:
23
    tech_keywords = ["代码", "编程", "API", "框架", "数据库", "算法"]
24
    return any(kw in input_dict["question"] for kw in tech_keywords)
25

26
# 构建条件路由
27
branch = RunnableBranch(
28
    (is_tech_question, tech_chain),   # 条件为 True 时走技术链
29
    general_chain                      # 默认走通用链
30
)
31

32
print(branch.invoke({"question": "如何优化数据库查询？"}))
33
# → 技术专家回答
34

35
print(branch.invoke({"question": "今天天气怎么样？"}))
36
# → 通用助手回答

1
# pip install langchain langchain-openai
2

3
from langchain_core.runnables import RunnableLambda
4
from langchain_core.prompts import ChatPromptTemplate
5
from langchain_core.output_parsers import StrOutputParser
6
from langchain_openai import ChatOpenAI
7

8
model = ChatOpenAI(model="gpt-4o-mini")
9

10
# 各领域处理链
11
math_chain = (
12
    ChatPromptTemplate.from_template("你是数学专家。请解答: {question}")
13
    | model | StrOutputParser()
14
)
15

16
history_chain = (
17
    ChatPromptTemplate.from_template("你是历史学家。请解答: {question}")
18
    | model | StrOutputParser()
19
)
20

21
general_chain = (
22
    ChatPromptTemplate.from_template("你是通用助手。请回答: {question}")
23
    | model | StrOutputParser()
24
)
25

26
# 路由表：分类标签 → 对应的 Runnable
27
route_map = {
28
    "math":    math_chain,
29
    "history": history_chain,
30
    "general": general_chain,
31
}
32

33
# 第一步：LLM 分类器，输出路由标签
34
classifier_prompt = ChatPromptTemplate.from_template(
35
    "请判断以下问题属于哪个类别，只输出一个单词：math / history / general\n"
36
    "问题: {question}"
37
)
38
classifier = classifier_prompt | model | StrOutputParser()
39

40
# 第二步：根据标签查表分发
41
def route(input_dict: dict) -> str:
42
    label = classifier.invoke(input_dict).strip().lower()
43
    # 未知标签兜底到 general
44
    return route_map.get(label, general_chain).invoke(input_dict)
45

46
router_chain = RunnableLambda(route)
47

48
print(router_chain.invoke({"question": "微积分的基本定理是什么？"}))
49
# → 数学专家回答
50

51
print(router_chain.invoke({"question": "第一次世界大战的导火索是什么？"}))
52
# → 历史学家回答

1
输入: {question: ...}
2
        │
3
        ▼
4
  LLM 分类器
5
  (classifier_prompt | model | StrOutputParser)
6
        │
7
        ▼ label: "math" / "history" / "general"
8
        │
9
  route_map 查表
10
        │
11
   ┌────┴────┬──────────┐
12
   ▼         ▼          ▼
13
math_chain  history_chain  general_chain
14
   └────┬────┴──────────┘
15
        ▼
16
     最终输出

1
# pip install langchain langchain-openai
2

3
from langchain_openai import ChatOpenAI
4
from langchain_core.callbacks import StdOutCallbackHandler
5

6
model = ChatOpenAI(model="gpt-4o-mini")
7

8
# 方式 1：在 invoke 时传入回调
9
result = model.invoke(
10
    "你好",
11
    config={"callbacks": [StdOutCallbackHandler()]}
12
)
13

14
# 方式 2：在构造时绑定回调
15
model_with_cb = ChatOpenAI(
16
    model="gpt-4o-mini",
17
    callbacks=[StdOutCallbackHandler()]
18
)

1
# pip install langchain langchain-openai
2

3
from langchain_core.callbacks import BaseCallbackHandler
4
from langchain_core.outputs import LLMResult
5
from langchain_openai import ChatOpenAI
6
from typing import Any
7

8

9
class TokenCounterCallback(BaseCallbackHandler):
10
    """自定义回调：统计 Token 消耗"""
11

12
    def __init__(self):
13
        self.total_prompt_tokens = 0
14
        self.total_completion_tokens = 0
15
        self.total_cost = 0.0
16
        self.call_count = 0
17

18
    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
19
        """LLM 调用结束时触发"""
20
        self.call_count += 1
21

22
        # 从 response 中提取 token 使用信息
23
        if response.llm_output:
24
            token_usage = response.llm_output.get("token_usage", {})
25
            prompt_tokens = token_usage.get("prompt_tokens", 0)
26
            completion_tokens = token_usage.get("completion_tokens", 0)
27

28
            self.total_prompt_tokens += prompt_tokens
29
            self.total_completion_tokens += completion_tokens
30

31
            print(f"\n--- 第 {self.call_count} 次调用 ---")
32
            print(f"  Prompt tokens:     {prompt_tokens}")
33
            print(f"  Completion tokens: {completion_tokens}")
34
            print(f"  Total tokens:      {prompt_tokens + completion_tokens}")
35

36
    def report(self) -> str:
37
        """打印累计统计报告"""
38
        total = self.total_prompt_tokens + self.total_completion_tokens
39
        return (
40
            f"\n===== Token 使用统计 =====\n"
41
            f"总调用次数:       {self.call_count}\n"
42
            f"总 Prompt tokens:  {self.total_prompt_tokens}\n"
43
            f"总 Completion tokens: {self.total_completion_tokens}\n"
44
            f"总 Token 数:       {total}\n"
45
        )
46

47

48
# 使用示例
49
counter = TokenCounterCallback()
50
model = ChatOpenAI(model="gpt-4o-mini")
51

52
# 多次调用
53
model.invoke("什么是 LCEL？", config={"callbacks": [counter]})
54
model.invoke("什么是 Runnable？", config={"callbacks": [counter]})
55

56
# 打印累计报告
57
print(counter.report())

1
# pip install langchain langchain-openai
2

3
from langchain_core.prompts import ChatPromptTemplate
4
from langchain_core.output_parsers import StrOutputParser
5
from langchain_openai import ChatOpenAI
6

7
chain = (
8
    ChatPromptTemplate.from_template("解释 {topic}")
9
    | ChatOpenAI(model="gpt-4o-mini")
10
    | StrOutputParser()
11
)
12

13
# counter 会收到链中每个步骤的事件
14
counter = TokenCounterCallback()
15
chain.invoke({"topic": "Transformer"}, config={"callbacks": [counter]})

1
# pip install langchain-core
2

3
from langchain_core.messages import (
4
    SystemMessage,
5
    HumanMessage,
6
    AIMessage,
7
    ToolMessage,
8
)
9

10
messages = [
11
    SystemMessage(content="你是一个有帮助的助手。"),
12
    HumanMessage(content="北京今天天气怎么样？"),
13
    AIMessage(
14
        content="",
15
        additional_kwargs={
16
            "tool_calls": [{
17
                "id": "call_abc123",
18
                "type": "function",
19
                "function": {
20
                    "name": "get_weather",
21
                    "arguments": '{"city": "北京"}'
22
                }
23
            }]
24
        }
25
    ),
26
    ToolMessage(
27
        content='{"temp": 22, "condition": "晴"}',
28
        tool_call_id="call_abc123"
29
    ),
30
    AIMessage(content="北京今天天气晴朗，气温约 22 度。"),
31
]

1
# pip install langchain langchain-openai
2

3
from langchain_core.prompts import ChatPromptTemplate
4
from langchain_core.output_parsers import StrOutputParser
5
from langchain_openai import ChatOpenAI
6

7
chain = (
8
    ChatPromptTemplate.from_template("翻译成英文: {text}")
9
    | ChatOpenAI(model="gpt-4o-mini")
10
    | StrOutputParser()
11
)
12

13
# 查看链的输入输出 Schema
14
print(chain.input_schema.model_json_schema())
15
# {'properties': {'text': {'title': 'Text', 'type': 'string'}},
16
#  'required': ['text'], 'title': 'PromptInput', 'type': 'object'}
17

18
print(chain.output_schema.model_json_schema())
19
# {'title': 'StrOutputParserOutput', 'type': 'string'}