1
# 构建消息列表
2
messages = [SystemMessage(content="你叫小亮，是一个乐于助人的人工助手"),
3
            HumanMessage(content="你是谁")
4
            ]
5
# 同步调用大模型
6
response = model.invoke(messages)
7
# 打印结果
8
print(response.content)
9
print(type(response))

1
response = model.invoke(messages)

1
'''流式输出'''
2
response = model.stream(messages)
3
for chunk in response:
4
    print(chunk.content, end="",flush=True) # 刷新缓冲区 (无换行符，缓冲区未刷新，内容可能不会立即显示)
5
print("\n")
6
print(type(response))

1
# 批量调用大模型
2
response = model.batch(batch_input)
3
for q, r in zip(questions, response):
4
    print(f"问题：{q}\n回答：{r}\n")

1
async def main():
2
    # 异步调用一条请求
3
    response = await model.ainvoke("解释一下LangChain是什么")
4
    print(response)
5

6
# 运行异步程序的入口点
7
asyncio.run(main())

1
# 创建一个PromptTemplate对象，用于生成格式化的提示词模板
2
# 该模板包含两个变量：role（角色）和question（问题）
3
template = PromptTemplate(template="你是一个专业的{role}工程师，请回答我的问题给出回答，我的问题是：{question}",
4
                        input_variables=['role', 'question'])
5

6
# 使用模板格式化具体的提示词内容
7
# 将role替换为"python开发"，question替换为"冒泡排序怎么写？"
8
prompt = template.format(role="python开发",question="冒泡排序怎么写？")
9

10
# 输出格式化后的提示词内容
11
print(prompt)

1
# 创建一个PromptTemplate对象，用于生成格式化的提示词模板
2
# 模板包含两个占位符：{role}表示角色，{question}表示问题
3
template = PromptTemplate.from_template("你是一个专业的{role}工程师，请回答我的问题给出回答，我的问题是：{question}")
4

5
# 使用指定的角色和问题参数来格式化模板，生成最终的提示词字符串
6
# role: 工程师角色描述
7
# question: 具体的技术问题
8
prompt = template.format(role="python开发",question="冒泡排序怎么写？")
9

10
# 输出生成的提示词
11
print(prompt)

1
# 创建一个包含时间变量的模板，时间变量使用partial_variables预设为当前时间
2
# 然后格式化问题生成最终提示词
3
template1 = PromptTemplate.from_template("现在时间是：{time},请对我的问题给出答案，我的问题是：{question}",
4
partial_variables={"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
5
prompt1 = template1.format(question="今天是几号？")
6
print(prompt1)
7

8
# 创建一个包含时间变量的模板，通过partial方法预设时间变量为当前时间
9
# 然后格式化问题生成最终提示词
10
template2 = PromptTemplate.from_template("现在时间是：{time},请对我的问题给出答案，我的问题是：{question}")
11
partial = template2.partial(time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
12
prompt2 = partial.format(question="今天是几号？")
13
print(prompt2)

1
# 创建一个PromptTemplate模板，用于生成介绍某个主题的提示词
2
# 该模板包含两个占位符：topic（主题）和length（字数限制）
3
template1 = PromptTemplate.from_template("请用一句话介绍{topic}，要求通俗易懂\n") + "内容不超过{length}个字"
4
# 使用format方法填充模板中的占位符，生成具体的提示词
5
prompt1 = template1.format(topic="LangChain", length=20)
6
print(prompt1)
7

8
# 分别创建两个独立的PromptTemplate模板
9
prompt_a = PromptTemplate.from_template("请用一句话介绍{topic}，要求通俗易懂\n")
10
prompt_b = PromptTemplate.from_template("内容不超过{length}个字")
11
# 将两个模板进行拼接组合
12
prompt_all = prompt_a + prompt_b
13
# 填充组合后模板的占位符，生成最终的提示词
14
prompt2 = prompt_all.format(topic="LangChain", length=20)
15
print(prompt2)

1
# 创建聊天提示模板，包含系统角色设定、用户询问和AI回答的对话历史
2
# 以及用户当前输入的占位符
3
prompt_template = ChatPromptTemplate([
4
    ("system", "你是一个AI助手，你的名字是{name}"),
5
    ("human", "你能做什么事"),
6
    ("ai", "我可以陪你聊天，讲笑话，写代码"),
7
    ("human", "{user_input}"),
8
])
9

10
# 使用指定的参数格式化提示模板，生成最终的提示字符串
11
# name: AI助手的名称
12
# user_input: 用户的当前输入
13
prompt = prompt_template.format(name="小张", user_input="你可以做什么")
14
print(prompt)

1
# 创建聊天提示模板，包含系统角色设定和用户问题格式
2
# 系统消息定义了AI的角色，人类消息定义了问题的输入格式
3
chat_prompt = ChatPromptTemplate.from_messages([
4
    ("system", "你是一个{role}，请回答我提出的问题"),
5
    ("human", "请回答:{question}")
6
])
7

8
# 使用指定的角色和问题参数填充模板，生成具体的提示内容
9
# role: 指定AI扮演的角色
10
# question: 用户提出的具体问题
11
prompt_value = chat_prompt.invoke({"role": "python开发工程师", "question": "冒泡排序怎么写"})
12

13
# 输出生成的提示内容
14
print(prompt_value.to_string())

1
# 构建一个 ChatPromptTemplate，包含多种消息类型：
2
prompt = ChatPromptTemplate.from_messages([
3
    # 插入 memory 占位符，用于填充历史对话记录（如多轮对话上下文）
4
    MessagesPlaceholder("memory"),
5

6
    # 添加一条系统消息，设定 AI 的角色或行为准则
7
    SystemMessage("你是一个资深的Python应用开发工程师，请认真回答我提出的Python相关的问题"),
8

9
    # 添加一条用户问题消息，用变量 {question} 表示
10
    ("human", "{question}")
11
])
12

13
# 调用 prompt.invoke 来格式化整个 Prompt 模板
14
# 传入的参数中：
15
# - memory：是一组历史消息，表示之前的对话内容（多轮上下文）
16
# - question：是当前用户的问题
17
prompt_value = prompt.invoke({
18
    "memory": [
19
        # 用户第一轮说的话
20
        HumanMessage("我的名字叫亮仔，是一名程序员"),
21
        # AI 第一轮的回应
22
        AIMessage("好的，亮仔你好")
23
    ],
24
    # 当前问题：结合上下文，测试模型是否记住了用户名字
25
    "question": "请问我的名字叫什么？"
26
})
27

28
# 打印生成的完整 prompt 文本，格式化后的聊天记录
29
print(prompt_value.to_string())

1
# 使用 ChatPromptTemplate 构建一个多角色对话提示模板
2
prompt = ChatPromptTemplate.from_messages([
3
    # 占位符，用于插入对话“记忆”内容，即之前的聊天记录（历史上下文）
4
    ("placeholder", "{memory}"),
5

6
    # 系统消息，用于设定 AI 的角色 —— 是一个资深的 Python 应用开发工程师
7
    SystemMessage("你是一个资深的Python应用开发工程师，请认真回答我提出的Python相关的问题"),
8

9
    # 用户当前提问，使用变量 {question} 进行动态填充
10
    ("human", "{question}")
11
])
12

13
# 使用 invoke 方法传入上下文变量，生成格式化后的对话 prompt 内容
14
prompt_value = prompt.invoke({
15
    # memory：是之前的对话上下文，会被插入到 {memory} 的位置
16
    "memory": [
17
        # 用户第一轮对话
18
        HumanMessage("我的名字叫亮仔，是一名程序员"),
19
        # AI 第一轮回答
20
        AIMessage("好的，亮仔你好")
21
    ],
22

23
    # 当前的问题，将替换模板中的 {question}
24
    "question": "请问我的名字叫什么？"
25
})
26
# 使用 .to_string() 将格式化后的对话链转换成纯文本字符串，方便查看输出
27
print(prompt_value.to_string())

1
# 创建字符串输出解析器，用于解析模型返回的结果
2
parser = StrOutputParser ()
3

4
# 打印解析后的结构化结果
5
response = parser.invoke(result)

1
chat_prompt = ChatPromptTemplate.from_messages([
2
    ("system", "你是一个{role}，请简短回答我提出的问题，结果返回json格式，q字段表示问题，a字段表示答案。"),
3
    ("human", "请回答:{question}")
4
])
5

6

7
# 创建JSON输出解析器实例
8
parser = JsonOutputParser()
9
# 调用解析器处理结果数据，将输入转换为JSON格式的响应
10
response = parser.invoke(result)

1
class Person(BaseModel):
2
    time: str = Field(description="时间")
3
    person: str = Field(description="人物")
4
    event: str = Field(description="事件")
5

6
 # 创建JSON输出解析器，用于将model输出解析为Person对象
7
parser = JsonOutputParser(pydantic_object=Person)
8

9
# 获取格式化指令，告诉model如何输出符合要求的JSON格式
10
format_instructions = parser.get_format_instructions()
11
# 格式化提示词，填入具体主题和格式化指令
12
prompt = chat_prompt.format_messages(topic="小米", format_instructions=format_instructions)
13

14
# 初始化ChatOllama大语言模型实例
15
model = ChatOllama(model="qwen3:14b", reasoning=False)
16

17
# 调用大语言模型获取响应结果
18
result = model.invoke(prompt)
19

20
# 使用解析器将模型输出解析为结构化数据
21
response = parser.invoke(result)

1
# 创建逗号分隔列表输出解析器实例
2
parser = CommaSeparatedListOutputParser()
3

4
# 获取格式化指令，用于指导模型输出格式
5
format_instructions = parser.get_format_instructions()
6

7
# 格式化聊天提示消息，将占位符替换为实际值
8
prompt = chat_prompt.format_messages(topic="小米", format_instructions=format_instructions)

1
# 创建 XML 输出解析器实例
2
parser = XMLOutputParser()
3

4
# 获取格式化指令（这会告诉模型如何以 XML 格式输出）
5
format_instructions = parser.get_format_instructions()
6
# 创建提示模板
7
chat_prompt = ChatPromptTemplate.from_messages([
8
    ("system", f"你是一个AI助手，只能输出XML格式的结构化数据。{format_instructions}"),
9
    ("human", "请生成5个关于{topic}的内容，每个内容包含<name>和<description>两个字段")
10
])
11

12
# 格式化提示，将 {topic} 替换为实际主题
13
prompt = chat_prompt.format_messages(topic="小米", format_instructions=format_instructions)

1
class Product(BaseModel):
2
    name: str = Field(description="产品名称")
3
    category: str = Field(description="产品类别")
4
    description: str = Field(description="产品简介")
5

6
    @field_validator("description")
7
    def validate_description(cls, value):
8

9
        if len(value) < 10:
10
            raise ValueError('产品简介长度必须大于等于10')
11
        return value
12

13

14
# 创建Pydantic输出解析器实例，用于解析模型输出为Product对象
15
parser = PydanticOutputParser(pydantic_object=Product)
16

17
# 获取格式化指令，用于指导模型输出符合Product模型的JSON格式
18
format_instructions = parser.get_format_instructions()
19

20
# 创建聊天提示模板，包含系统消息和人类消息
21
prompt_template = ChatPromptTemplate.from_messages([
22
    ("system", "你是一个AI助手，你只能输出结构化的json数据\n{format_instructions}"),
23
    ("human", "请你输出标题为：{topic}的新闻内容")
24
])
25

26
# 格式化提示消息，填充主题和格式化指令
27
prompt = prompt_template.format_messages(topic="小米", format_instructions=format_instructions)

1
prompt | model | parser

1
#=========================================================
2
#初始化模型，提示词模板， 解析器
3
'''初始化 模型'''
4
model = ChatOpenAI(
5
    model=MODEL,
6
    temperature=0,
7
    max_tokens=None,
8
    timeout=None,
9
    max_retries=2,
10
    base_url=DEEPSEEK_BASE_URL,
11
    api_key=DEEPSEEK_API_KEY,
12
)
13

14
'''初始化提示词模板'''
15
# 创建聊天提示模板，包含系统角色设定和用户问题输入
16
chat_prompt = ChatPromptTemplate.from_messages([
17
    ("system", "你是一个{role}，请简短回答我提出的问题"),
18
    ("human", "请回答:{question}")
19
])
20

21
'''初始化解析器'''
22
parser = StrOutputParser ()

1
#=======================================================================
2
#调用
3
#提示词调用
4
chat_prompt = ChatPromptTemplate.from_messages([
5
    ("system", "你是一个{role}，请简短回答我提出的问题"),
6
    ("human", "请回答:{question}")
7
])
8
prompt = chat_prompt.invoke({"role": "AI助手", "question": "什么是LangChain"})
9

10
#模型调用
11
result = model.invoke(prompt)
12

13
#解析器调用
14
response = parser.invoke(result)

1
# 构建处理链：提示模板 -> 模型 -> 输出解析器
2
chain = chat_prompt | model | parser
3

4
# 执行处理链并记录最终结果及数据类型
5
result_chain = chain.invoke({"role": "AI助手", "question": "什么是LangChain"})
6
logger.info(f"Chain执行结果:\n {result_chain}")
7
logger.info(f"Chain执行结果类型: {type(result_chain)}")

1
# 构建提示词
2
english_prompt = ChatPromptTemplate.from_messages([
3
    ("system", "你是一个英语翻译专家，你叫小英"),
4
    ("human", "{query}")
5
])
6

7
japanese_prompt = ChatPromptTemplate.from_messages([
8
    ("system", "你是一个日语翻译专家，你叫小日"),
9
    ("human", "{query}")
10
])
11

12
korean_prompt = ChatPromptTemplate.from_messages([
13
    ("system", "你是一个韩语翻译专家，你叫小韩"),
14
    ("human", "{query}")
15
])
16

17

18
def determine_language(inputs):
19
    """判断语言种类"""
20
    query = inputs["query"]
21
    if "日语" in query:
22
        return "japanese"
23
    elif "韩语" in query:
24
        return "korean"
25
    else:
26
        return "english"
27

28

29
# 初始化Ollama聊天模型，指定使用qwen3:8b模型，关闭推理模式
30
model = ChatOllama(model="qwen3:8b", reasoning=False)
31
# 创建字符串输出解析器，用于处理模型输出
32
parser = StrOutputParser()
33
# 创建一个可运行的分支链，根据输入文本的语言类型选择相应的处理流程
34
# 返回值：
35
#   RunnableBranch对象，可根据输入动态选择执行路径的可运行链
36
chain = RunnableBranch(
37
    (lambda x: determine_language(x) == "japanese", japanese_prompt | model | parser),
38
    (lambda x: determine_language(x) == "korean", korean_prompt | model | parser),
39
    (english_prompt | model | parser)
40
)
41

42
# 测试查询
43
test_queries = [
44
    {'query': '请你用韩语翻译这句话:"见到你很高兴"'},
45
    {'query': '请你用日语翻译这句话:"见到你很高兴"'},
46
    {'query': '请你用英语翻译这句话:"见到你很高兴"'}
47
]
48

49
for query_input in test_queries:
50
    # 判断使用哪个提示词
51
    lang = determine_language(query_input)
52
    logger.info(f"检测到语言类型: {lang}")
53

54
    # 根据语言类型选择对应的提示词并格式化
55
    if lang == "japanese":
56
        prompt = japanese_prompt
57
    elif lang == "korean":
58
        prompt = korean_prompt
59
    else:
60
        prompt = english_prompt
61

62
    # 格式化提示词并打印
63
    formatted_messages = prompt.format_messages(**query_input)
64
    logger.info("格式化后的提示词:")
65
    for msg in formatted_messages:
66
        logger.info(f"[{msg.type}]: {msg.content}")
67

68
    # 执行链
69
    result = chain.invoke(query_input)
70
    logger.info(f"输出结果: {result}\n")

1
# 设置本地模型，不使用深度思考
2
model = ChatOllama(model="qwen3:8b", reasoning=False)
3

4
# 子链1提示词
5
prompt1 = ChatPromptTemplate.from_messages([
6
    ("system", "你是一个知识渊博的计算机专家，请用中文简短回答"),
7
    ("human", "请简短介绍什么是{topic}")
8
])
9
# 子链1解析器
10
parser1 = StrOutputParser()
11
# 子链1：生成内容
12
chain1 = prompt1 | model | parser1
13

14
# 子链2提示词
15
prompt2 = ChatPromptTemplate.from_messages([
16
    ("system", "你是一个翻译助手，将用户输入内容翻译成英文"),
17
    ("human", "{input}")
18
])
19
# 子链2解析器
20
parser2 = StrOutputParser()
21

22
# 子链2：翻译内容
23
chain2 = prompt2 | model | parser2
24

25
# 组合成一个复合 Chain，使用 lambda 函数将chain1执行结果content内容添加input键作为参数传递给chain2
26
full_chain = chain1 | (lambda content: {"input": content}) | chain2
27

28
# 调用复合链
29
result = full_chain.invoke({"topic": "langchain"})
30
logger.info(result)

1
# 设置本地模型，不使用深度思考
2
model = ChatOllama(model="qwen3:8b", reasoning=False)
3

4
# 并行链1提示词
5
prompt1 = ChatPromptTemplate.from_messages([
6
    ("system", "你是一个知识渊博的计算机专家，请用中文简短回答"),
7
    ("human", "请简短介绍什么是{topic}")
8
])
9
# 并行链1解析器
10
parser1 = StrOutputParser()
11
# 并行链1：生成中文结果
12
chain1 = prompt1 | model | parser1
13

14
# 并行链2提示词
15
prompt2 = ChatPromptTemplate.from_messages([
16
    ("system", "你是一个知识渊博的计算机专家，请用英文简短回答"),
17
    ("human", "请简短介绍什么是{topic}")
18
])
19
# 并行链2解析器
20
parser2 = StrOutputParser()
21

22
# 并行链2：生成英文结果
23
chain2 = prompt2 | model | parser2
24

25
# 创建并行链,用于同时执行多个语言处理链
26
parallel_chain = RunnableParallel({
27
    "chinese": chain1,
28
    "english": chain2
29
})
30

31
# 调用复合链
32
result = parallel_chain.invoke({"topic": "langchain"})
33
logger.info(result)

1
# 设置本地模型，不使用深度思考
2
model = ChatOllama(model="qwen3:8b", reasoning=False)
3

4

5
# 一个简单的打印函数，调试用
6
def debug_print(x):
7
    logger.info(f"中间结果:{x}")
8
    return {"input": x}
9

10

11
# 子链1提示词
12
prompt1 = ChatPromptTemplate.from_messages([
13
    ("system", "你是一个知识渊博的计算机专家，请用中文简短回答"),
14
    ("human", "请简短介绍什么是{topic}")
15
])
16
# 子链1解析器
17
parser1 = StrOutputParser()
18
# 子链1：生成内容
19
chain1 = prompt1 | model | parser1
20

21
# 子链2提示词
22
prompt2 = ChatPromptTemplate.from_messages([
23
    ("system", "你是一个翻译助手，将用户输入内容翻译成英文"),
24
    ("human", "{input}")
25
])
26
# 子链2解析器
27
parser2 = StrOutputParser()
28

29
# 子链2：翻译内容
30
chain2 = prompt2 | model | parser2
31
# 创建一个可运行的调试节点，用于打印中间结果
32
debug_node = RunnableLambda(debug_print)
33

34
# 构建完整的处理链，将chain1、调试打印和chain2串联起来
35
full_chain = chain1 | debug_print | chain2
36

37
# 调用复合链
38
result = full_chain.invoke({"topic": "langchain"})
39
logger.info(f"最终结果:{result}")

1
def retrieval_doc(question):
2
    """模拟知识库检索"""
3
    logger.info(f"检索器接收到用户提出问题：{question}")
4
    return "你是一个说话风趣幽默的AI助手，你叫亮仔"
5

6

7
# 设置本地模型，不使用深度思考
8
model = ChatOllama(model="qwen3:8b", reasoning=False)
9

10
# 构建提示词
11
prompt = ChatPromptTemplate.from_messages([
12
    ("system", "{retrieval_info}"),
13
    ("human", "请简短回答{question}")
14
])
15
# 创建字符串输出解析器
16
parser = StrOutputParser()
17
# 构建完整链条（Chain）：
18
# - 首先从输入中取出 question（问题）并传给两个函数：
19
#   1. 传给 lambda 获取 retrieval_info（角色设定）
20
#   2. 使用 itemgetter 保留 question 原文
21
# - 然后将这些内容输入 prompt 模板
22
# - 模型执行推理
23
# - 最后解析模型输出为纯文本
24
chain = {
25
            "retrieval_info": lambda x: retrieval_doc(x["question"]),
26
            "question": itemgetter("question")
27
        } | prompt | model | parser
28

29
# 5.执行链
30
result = chain.invoke({'question': '你是谁，什么叫LangChain？'})
31
logger.info(result)

1
def retrieval_doc(question):
2
    """模拟知识库检索"""
3
    logger.info(f"检索器接收到用户提出问题：{question}")
4
    return "你是一个说话风趣幽默的AI助手，你叫亮仔"
5

6

7
# 设置本地模型，不使用深度思考
8
model = ChatOllama(model="qwen3:8b", reasoning=False)
9

10
# 构建提示词
11
prompt = ChatPromptTemplate.from_messages([
12
    ("system", "{retrieval_info}"),
13
    ("human", "请简短回答{question}")
14
])
15
# 创建字符串输出解析器
16
parser = StrOutputParser()
17
# 构建链
18
# 1. 使用 RunnablePassthrough.assign 注入 retrieval_info 字段，
19
#    实际上是让 `retrieval_doc` 函数在链开始时执行，并将其结果加到 inputs 字典中。
20
#    即：输入 {"question": "xxx"} -> 输出 {"question": "xxx", "retrieval_info": "你是一个愤怒的语文老师..."}
21
# 2. 该完整字典被传入 prompt 中生成对话消息
22
# 3. 然后传入 model 获取回答
23
# 4. 最后使用 parser 提取字符串输出
24
chain = RunnablePassthrough.assign(retrieval_info=retrieval_doc) | prompt | model | parser
25

26
# 执行链
27
result = chain.invoke({'question': '你是谁，什么是LangChain'})
28
logger.info(result)

1
# 设置本地模型，不使用深度思考
2
model = ChatOllama(model="qwen3:8b", reasoning=False)
3

4
# 并行链1提示词
5
prompt1 = ChatPromptTemplate.from_messages([
6
    ("system", "你是一个知识渊博的计算机专家，请用中文简短回答"),
7
    ("human", "请简短介绍什么是{topic}")
8
])
9
# 并行链1解析器
10
parser1 = StrOutputParser()
11
# 并行链1：生成中文结果
12
chain1 = prompt1 | model | parser1
13

14
# 并行链2提示词
15
prompt2 = ChatPromptTemplate.from_messages([
16
    ("system", "你是一个知识渊博的计算机专家，请用英文简短回答"),
17
    ("human", "请简短介绍什么是{topic}")
18
])
19
# 并行链2解析器
20
parser2 = StrOutputParser()
21

22
# 并行链2：生成英文结果
23
chain2 = prompt2 | model | parser2
24

25
# 创建并行链,用于同时执行多个语言处理链
26
parallel_chain = RunnableParallel({
27
    "chinese": chain1,
28
    "english": chain2
29
})
30
# 将并行链的计算图绘制为PNG图片并保存
31
# parallel_chain.get_graph().draw_png("chain.png")
32
# 打印并行链的ASCII图形表示
33
parallel_chain.get_graph().print_ascii()
34
# 调用复合链
35
result = parallel_chain.invoke({"topic": "langchain"})
36
logger.info(f"最终结果:{result}")

1
# 定义 Prompt
2
prompt = ChatPromptTemplate.from_messages([
3
    MessagesPlaceholder(variable_name="history"),  # 用于插入历史消息
4
    ("human", "{input}")
5
])
6
# 初始化Ollama语言模型实例，配置基础URL、模型名称和推理模式
7
llm = ChatOllama(model="qwen3:14b", reasoning=False)
8
parser = StrOutputParser()
9
# 构建处理链：将提示词模板、语言模型和输出解析器组合
10
chain = prompt | llm | parser
11

12

13
#================================================================
14
# 创建内存聊天历史记录实例，用于存储对话历史
15
history = InMemoryChatMessageHistory()
16
# 创建带消息历史的可运行对象，用于处理带历史记录的对话
17
runnable = RunnableWithMessageHistory(
18
    chain,
19
    get_session_history=lambda session_id: history,
20
    input_messages_key="input",  # 指定输入键
21
    history_messages_key="history"  # 指定历史消息键
22
)
23
# 清空历史记录
24
history.clear()
25
# 配置运行时参数，设置会话ID
26
config = RunnableConfig(configurable={"session_id": "default"})
27
logger.info(runnable.invoke({"input": "我叫什么？我的爱好是什么？"}, config))

1
# 存储会话历史
2
store = {}
3
# 保留的历史轮数
4
k = 2
5
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
6
    """获取或创建会话历史"""
7
    if session_id not in store:
8
        store[session_id] = InMemoryChatMessageHistory()
9

10
    # 只保留最近 k 轮对话（2k 条消息）
11
    history = store[session_id]
12
    if len(history.messages) > k * 2:
13
        # 保留最近的消息
14

15
        # 手动修剪
16
        # trimmed_messages = history.messages[-k * 2:]
17

18
        # 使用trim_message进行修建
19
        trimmed_messages = trim_messages(
20
            history.messages,
21
            max_tokens=k * 2,  # 保留的消息数量
22
            token_counter=llm,
23
            strategy="last",    # 保留最近的
24
            include_system=False  # 是否包含系统消息
25
        )
26

27

28
        history.clear()
29
        history.add_messages(trimmed_messages)
30

31
    return history
32

33
# 创建带历史的链
34
chain = RunnableWithMessageHistory(
35
    prompt | llm,
36
    get_session_history,
37
    input_messages_key="question",
38
    history_messages_key="history"
39
)
40

41
# 配置
42
config = RunnableConfig(configurable={"session_id": "demo"})
43

44
# 主循环
45
print("开始对话（输入 'quit' 退出）")
46
while True:
47
    question = input("\n输入问题：")
48
    if question.lower() in ['quit', 'exit', 'q']:
49
        break
50

51
    response = chain.invoke({"question": question}, config)
52
    print("AI回答:", response.content)
53

54
    # 可选：显示当前历史消息数
55
    history = get_session_history("demo")
56
    print(f"[当前历史消息数: {len(history.messages)}]")

1
# 1. 按消息数量裁剪（保留最近3条）
2
trimmed = trim_messages(
3
    messages,
4
    max_tokens=3,  # 保留3条消息
5
    strategy="last",
6
    include_system=True  # 保留系统消息
7
)
8

9
# 2. 按token数量裁剪（更精细）
10
trimmed = trim_messages(
11
    messages,
12
    max_tokens=100,  # 保留最近100个token
13
    strategy="last",
14
    token_counter=llm  # 使用模型的tokenizer计数
15
)
16

17
# 3. 保留第一条消息（如系统提示）
18
trimmed = trim_messages(
19
    messages,
20
    max_tokens=100,
21
    strategy="last",
22
    start_on="human",  # 从第一条人类消息开始
23
    include_system=True
24
)

1
# Redis 配置
2
REDIS_URL = "redis://localhost:6379/0"
3

4
# 初始化模型
5
llm = ChatOllama(model="qwen3:14b", reasoning=False)
6

7
# 创建提示模板
8
prompt = ChatPromptTemplate.from_messages([
9
    MessagesPlaceholder("history"),
10
    ("human", "{question}")
11
])
12

13
# 保留的历史轮数
14
k = 2
15

16

17
def get_session_history(session_id: str) -> RedisChatMessageHistory:
18
    """获取或创建会话历史（使用 Redis）"""
19
    # 创建 Redis 历史对象
20
    history = RedisChatMessageHistory(
21
        session_id=session_id,
22
        url=REDIS_URL,
23
        ttl=3600  # 1小时过期
24
    )
25

26
    # 自动修剪：只保留最近 k 轮对话（2k 条消息）
27
    if len(history.messages) > k * 2:
28
        # 保留最近的消息
29
        messages_to_keep = history.messages[-k * 2:]
30
        history.clear()
31
        history.add_messages(messages_to_keep)
32

33
    return history
34

35

36
# 创建带历史的链
37
chain = RunnableWithMessageHistory(
38
    prompt | llm,
39
    get_session_history,
40
    input_messages_key="question",
41
    history_messages_key="history"
42
)
43

44
# 配置
45
config = RunnableConfig(configurable={"session_id": "demo"})
46

47
# 主循环
48
print("开始对话（输入 'quit' 退出）")
49
while True:
50
    question = input("\n输入问题：")
51
    if question.lower() in ['quit', 'exit', 'q']:
52
        break
53

54
    response = chain.invoke({"question": question}, config)
55
    logger.info(f"AI回答:{response.content}")
56

57
    # 可选：显示当前历史消息数
58
    history = get_session_history("demo")
59
    logger.info(f"[当前历史消息数: {len(history.messages)}]")

1
@tool
2
def get_weather(loc):
3
    # Step 1.构建请求
4
    url = "https://api.openweathermap.org/data/2.5/weather"
5

6
    # Step 2.设置查询参数
7
    params = {
8
        "q": loc,
9
        "appid": os.getenv("OPENWEATHER_API_KEY"),  # 输入API key
10
        "units": "metric",  # 使用摄氏度而不是华氏度
11
        "lang": "zh_cn"  # 输出语言为简体中文
12
    }
13

14
    # Step 3.发送GET请求
15
    response = httpx.get(url, params=params)
16

17
    # Step 4.解析响应
18
    data = response.json()
19
    return json.dumps(data)
20

21

22
# 初始化ChatOllama语言模型实例，用于处理自然语言任务
23
llm = ChatOllama(model="qwen3:14b", reasoning=False)
24

25
# 创建聊天提示模板，定义agent的对话结构和角色
26
prompt = ChatPromptTemplate.from_messages(
27
    [
28
        ("system", "你是天气助手，请根据用户的问题，给出相应的天气信息"),
29
        ("human", "{input}"),
30
        ("placeholder", "{agent_scratchpad}"),
31
    ]
32
)
33

34
# 定义可用工具列表，包含获取天气信息的工具函数
35
tools = [get_weather]
36

37
# 创建工具调用agent，整合语言模型、工具和提示模板。该agent能够根据用户问题调用相应工具获取天气信息
38
agent = create_tool_calling_agent(llm, tools, prompt)
39

40
# 创建agent执行器，负责协调agent和工具的执行流程
41
# agent参数指定要执行的agent实例
42
# tools参数提供agent可调用的工具列表
43
# verbose参数设置为True，启用详细输出模式便于调试
44
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
45

46
# 执行agent，处理用户关于北京和上海天气的查询请求
47
result = agent_executor.invoke({"input": "请问今天北京和上海的天气怎么样，哪个城市更热？"})
48

49
# 输出执行结果
50
print(result)

1
@tool
2
def get_weather(loc):
3
    # Step 1.构建请求
4
    url = "https://api.openweathermap.org/data/2.5/weather"
5

6
    # Step 2.设置查询参数
7
    params = {
8
        "q": loc,
9
        "appid": os.getenv("OPENWEATHER_API_KEY"),  # 输入API key
10
        "units": "metric",  # 使用摄氏度而不是华氏度
11
        "lang": "zh_cn"  # 输出语言为简体中文
12
    }
13

14
    # Step 3.发送GET请求
15
    response = httpx.get(url, params=params)
16

17
    # Step 4.解析响应
18
    data = response.json()
19
    return json.dumps(data)
20

21

22
@tool
23
def write_file(content):
24
    print(f"写入文件内容：{content}")
25
    return "已成功写入本地文件。"
26

27

28
# 初始化ChatOllama语言模型实例，用于处理自然语言任务
29
llm = ChatOllama(model="qwen3:14b", reasoning=False)
30

31
# 创建聊天提示模板，定义系统角色和用户输入格式
32
prompt = ChatPromptTemplate.from_messages(
33
    [
34
        ("system",
35
         "你是天气助手，请根据用户的问题，给出相应的天气信息，如果用户需要将查询结果写入文件，请使用write_file工具"),
36
        ("human", "{input}"),
37
        ("placeholder", "{agent_scratchpad}"),
38
    ]
39
)
40

41
# 定义可用工具列表，包括获取天气信息和写入文件功能
42
tools = [get_weather, write_file]
43

44
# 创建工具调用代理，整合语言模型、工具和提示模板
45
agent = create_tool_calling_agent(llm, tools, prompt)
46

47
# 创建代理执行器，用于执行代理任务并管理工具调用过程
48
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
49

50
# 执行用户查询任务，获取北京和上海的温度信息并写入文件
51
result = agent_executor.invoke({"input": "查一下北京和上海现在的温度，并将结果写入本地的文件中。"})
52

53
print(result)

1
# 创建Google Serper API包装器实例
2
api_wrapper = GoogleSerperAPIWrapper()
3

4
# 创建Google搜索工具实例
5
search_tool = GoogleSerperRun(api_wrapper=api_wrapper)
6
# 将搜索工具添加到工具列表中
7
tools = [search_tool]
8

9
# 定义聊天提示模板，包含系统角色设定、用户输入和代理执行过程占位符
10
prompt = ChatPromptTemplate.from_messages([
11
    ("system", "你是一名助人为乐的助手，并且可以调用工具进行网络搜索，获取实时信息。"),
12
    ("human", "{input}"),
13
    ("placeholder", "{agent_scratchpad}"),
14
])
15

16
# 创建Ollama聊天模型实例，使用qwen3:8b模型
17
llm = ChatOllama(model="qwen3:8b", reasoning=False)
18

19
# 创建工具调用代理，整合语言模型、工具和提示模板
20
agent = create_tool_calling_agent(llm, tools, prompt)
21

22
# 创建代理执行器，用于执行代理并管理工具调用过程
23
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
24

25
# 调用代理执行器，传入用户查询问题
26
agent_executor.invoke({"input": "小米最近发布的新品是什么？"})