模型Models

初始化模型（本地模型）

from langchain_ollama import ChatOllama

# 1. 初始化 LLM
llm = ChatOllama(
    base_url="http://localhost:11434",
    model="qwen2.5:3b"
)

提示词Prompts

创建提示词模板+填充模板

from langchain.prompts import ChatPromptTemplate

# 创建提示词模板
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}助手"),
    ("human", "{query}")
])

# 填充模板
messages = prompt_template.format_messages(
    role="编程",
    query="如何用Python反转列表？"
)

链Chains

创建链+运行链

一个基础链👇

from langchain_ollama import ChatOllama
from langchain_classic.prompts import PromptTemplate
from langchain_classic.chains import LLMChain

# 1. 初始化模型
llm = ChatOllama(
    base_url="http://localhost:11434",
    model="qwen2.5:3b"
)

# 2. 创建提示词模板
template = """你是一个{expert_type}专家。
请回答以下问题：
问题：{question}
答案："""

prompt = PromptTemplate(
    input_variables=["expert_type", "question"],
    template=template
)

# 3. 创建链
chain = LLMChain(llm=llm, prompt=prompt)

# 4. 运行链
result = chain.run(
    expert_type="Python编程",
    question="什么是装饰器？"
)
print(result)

一个顺序链👇

from langchain_ollama import ChatOllama
from langchain_classic.prompts import PromptTemplate
from langchain_classic.chains import LLMChain
from langchain_classic.chains import SimpleSequentialChain

# 1. 初始化 LLM
llm = ChatOllama(
    base_url="http://localhost:11434",
    model="qwen2.5:3b"
)

# 第一个链：生成菜谱
recipe_template = """你是一位厨师，请为{ingredient}创建一道菜谱。"""
recipe_prompt = PromptTemplate(
    input_variables=["ingredient"],
    template=recipe_template
)
recipe_chain = LLMChain(llm=llm, prompt=recipe_prompt)

# 第二个链：生成购物清单
shopping_template = """根据以下菜谱，列出需要的购物清单：
菜谱：{recipe}
购物清单："""
shopping_prompt = PromptTemplate(
    input_variables=["recipe"],
    template=shopping_template
)
shopping_chain = LLMChain(llm=llm, prompt=shopping_prompt)

# 组合两个链
overall_chain = SimpleSequentialChain(
    chains=[recipe_chain, shopping_chain],
    verbose=True  # 显示执行过程
)

# 运行
result = overall_chain.run("鸡蛋")
print(result)

1. 链式调用 (Chaining)

• 将多个 LLM 任务按顺序连接起来

• 前一个任务的输出作为后一个任务的输入

• 展示了如何构建多步骤的AI工作流

2. SimpleSequentialChain 的使用

• 最简单的链式结构：A → B → C

• 每个链只接收上一个链的输出

• 适合线性、无分支的工作流程

3. PromptTemplate 的应用

• 如何创建可重用的提示词模板

• 使用 input_variables 动态插入内容

• 分离提示词逻辑与业务逻辑

4. LLMChain 的构建

• 将 LLM 与 PromptTemplate 组合成可执行单元

• 每个链都是一个独立的、可测试的组件

记忆Memory

1. ConversationBufferMemory - 基础记忆

特点：最简单的记忆类型，直接存储所有对话历史

from langchain_classic.memory import ConversationBufferMemory

memory = ConversationBufferMemory()
memory.save_context(
    {"input": "你好，我叫小明"},
    {"output": "你好小明！很高兴认识你。"}
)

# 继续对话
memory.save_context(
    {"input": "我今年25岁"},
    {"output": "25岁是很棒的年龄！"}
)

# 查看记忆内容
print(memory.buffer)

2. ConversationBufferWindowMemory - 滑动窗口记忆

特点：只保留最近 N 轮对话，防止上下文过长

from langchain_classic.memory import ConversationBufferWindowMemory

# 只保留最近3轮对话
memory = ConversationBufferWindowMemory(k=3)

memory.save_context({"input": "第一句话"}, {"output": "回复1"})
memory.save_context({"input": "第二句话"}, {"output": "回复2"})
memory.save_context({"input": "第三句话"}, {"output": "回复3"})
memory.save_context({"input": "第四句话"}, {"output": "回复4"})

# 只会记住第2、3、4轮对话
print(memory.buffer)

3. ConversationSummaryMemory - 摘要记忆

特点：不存储完整对话，而是存储摘要

from langchain_classic.memory import ConversationSummaryMemory
from langchain_ollama import ChatOllama

llm = ChatOllama(
    base_url="http://localhost:11434",
    model="qwen2.5:3b"
)
memory = ConversationSummaryMemory(llm=llm)

# 长对话会被自动总结
memory.save_context({"input": "我喜欢编程，特别是Python"}, {"output": "很棒！Python确实很受欢迎"})
memory.save_context({"input": "我平时也喜欢打篮球"}, {"output": "运动对健康很重要"})
memory.save_context({"input": "我还喜欢听音乐，尤其是古典音乐"}, {"output": "古典音乐很有品味"})

print(memory.buffer)
# 可能输出：用户喜欢编程（Python）、篮球和古典音乐

4. ConversationSummaryBufferMemory

特点：结合 buffer 和 summary，保留最近对话+较早摘要

5. ConversationKnowledgeGraphMemory

特点：将对话信息存储为知识图谱

6. VectorStoreRetrieverMemory

特点：将记忆存储在向量数据库中，基于语义检索

对比表格

Memory 类型	存储方式	优点	缺点	适用场景
BufferMemory	完整存储所有对话	信息完整，简单	上下文无限增长	短对话，需要完整历史
BufferWindowMemory	存储最近N轮	控制长度，防止过长	丢失早期信息	中等长度对话
SummaryMemory	存储摘要	节省token，处理长对话	可能丢失细节	超长对话，成本敏感
SummaryBufferMemory	最近完整+早期摘要	平衡细节与长度	实现较复杂	长短混合的对话
KnowledgeGraphMemory	知识图谱	支持推理，结构化	需要特定LLM能力	需要逻辑推理的场景
VectorStoreMemory	向量存储	语义检索，智能回忆	需要向量数据库	大规模记忆，智能推荐

选择建议

按对话长度选择：

• 短对话（<10轮）：ConversationBufferMemory

• 中长对话（10-30轮）：ConversationBufferWindowMemory 或 ConversationSummaryBufferMemory

• 超长对话：ConversationSummaryMemory

按功能需求选择：

• 需要精确回忆：Buffer 类型

• 需要总结归纳：Summary 类型

• 需要智能关联：VectorStore 类型

• 需要逻辑推理：KnowledgeGraph 类型

from langchain_ollama import ChatOllama
from langchain_classic.memory import ConversationBufferMemory
from langchain_classic.chains import ConversationChain

# 1. 初始化模型
llm = ChatOllama(
    base_url="http://localhost:11434",
    model="qwen2.5:3b"
)

# 2. 创建 Memory
memory = ConversationBufferMemory()

# 3. 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=False  # 关闭详细输出，避免混乱
)

# 4. 进行连续对话（关键：使用同一个 conversation 对象）
print("Human: 你好，我叫小明")
response1 = conversation.predict(input="你好，我叫小明")
print(f"AI: {response1}\n")

print("Human: 我来自北京")
response2 = conversation.predict(input="我来自北京")
print(f"AI: {response2}\n")

print("Human: 我刚才说我叫什么名字？从哪里来？")
response3 = conversation.predict(input="我刚才说我叫什么名字？从哪里来？")
print(f"AI: {response3}")

代理Agent

让AI使用工具

Agent = LLM大脑 + 工具(Tools) + 思考能力

from langchain_classic.agents import AgentType, initialize_agent
from langchain_classic.tools import Tool
from langchain_ollama import ChatOllama
import math

# 1. 定义工具
def calculator(input_str):
    """计算数学表达式"""
    try:
        # 安全地评估数学表达式
        result = eval(input_str, {"__builtins__": None}, {"math": math})
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {e}"

# 创建工具对象
calculator_tool = Tool(
    name="Calculator",
    func=calculator,
    description="用于计算数学表达式，例如：'3 * 4 + 5'"
)

# 2. 定义更多工具
def get_current_time(input_str=""):
    """获取当前时间"""
    from datetime import datetime
    return f"当前时间是: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

time_tool = Tool(
    name="CurrentTime",
    func=get_current_time,
    description="获取当前日期和时间"
)

# 3. 初始化Agent
llm = ChatOllama(
    base_url="http://localhost:11434",
    model="qwen2.5:3b"
)
tools = [calculator_tool, time_tool]

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,  # 最简单的Agent类型
    verbose=True,  # 显示思考过程
    handle_parsing_errors=True  # 处理解析错误
)

# 4. 运行Agent
print("=== Agent测试 ===")
result = agent.run("请计算 (25 * 4 + 18) / 2 等于多少？")
print(f"最终答案: {result}")

# 尝试更复杂的问题
result2 = agent.run("现在几点了？顺便计算一下圆周率乘以10的平方")

带记忆的agent

from langchain_ollama import ChatOllama
from langchain_classic.agents import initialize_agent, AgentType
from langchain_classic.memory import ConversationBufferMemory
from langchain_classic.tools import Tool

# 使用本地 Ollama 模型
llm = ChatOllama(
    model="qwen2.5:3b",
    temperature=0.3
)

# 创建记忆
memory = ConversationBufferMemory(memory_key="chat_history")

# 定义简单工具
def calculator(expression: str) -> str:
    try:
        result = eval(expression)
        return str(result)
    except:
        return "无法计算"

tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="用于数学计算，输入数学表达式如 '2+2'"
    ),
    Tool(
        name="Time",
        func=lambda x: "当前时间不可用（示例）",
        description="获取当前时间"
    )
]

# 创建带记忆的 Agent
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True,
    max_iterations=3
)

# 测试
print("测试带记忆的 Agent...")
print(agent.run("你好，我叫李四"))
print(agent.run("我刚才说我叫什么名字？"))
print(agent.run("15 + 27"))

简单RAG问答

文档加载、分块、向量化、检索

# 重点学习：文档加载、分块、向量化、检索
from langchain_community.document_loaders import PyPDFLoader
from langchain_classic.text_splitter import RecursiveCharacterTextSplitter
from langchain_ollama.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_classic.chains import RetrievalQA

from langchain_ollama import ChatOllama

# 完整RAG流程
loader = PyPDFLoader("2025.06六级真题第1套.pdf")
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)

embeddings = OllamaEmbeddings(
    model="qwen2.5:3b",  # 或你已下载的其他模型
)
vectorstore = Chroma.from_documents(chunks, embeddings)

llm = ChatOllama(
    base_url="http://localhost:11434",
    model="qwen2.5:3b"
)

# qa_chain 本身就是一个完整的链，它已经把 检索 + 问答 封装好了
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever()
)

# result = qa_chain.invoke("翻译题的主题是什么？")
# print("回答:", result["result"])

嵌入模型太大？内存不足

显示进度↓

from langchain_community.document_loaders import PyPDFLoader
from langchain_classic.text_splitter import RecursiveCharacterTextSplitter
from langchain_ollama.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_classic.chains import RetrievalQA
from langchain_ollama import ChatOllama
import time

print("🟢 开始执行 RAG 流程...")

# 1. 加载文档
try:
    print("1. 正在加载PDF文档...")
    loader = PyPDFLoader("2025.06六级真题第1套.pdf")
    documents = loader.load()
    print(f"   ✅ 加载成功！共 {len(documents)} 页")
    if len(documents) > 0:
        print(f"   第一页预览: {documents[0].page_content[:100]}...")
except Exception as e:
    print(f"   ❌ 加载失败: {e}")
    exit()

# 2. 文档分块
try:
    print("\n2. 正在分割文档...")
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    chunks = text_splitter.split_documents(documents)
    print(f"   ✅ 分割成功！共 {len(chunks)} 个文本块")
    if len(chunks) > 0:
        print(f"   第一个文本块: {chunks[0].page_content[:100]}...")
except Exception as e:
    print(f"   ❌ 分割失败: {e}")
    exit()

# 3. 向量化（检查Ollama服务）
try:
    print("\n3. 正在初始化嵌入模型...")
    print("   检查Ollama服务...")

    # 先测试Ollama连接
    import requests

    ollama_test = requests.get("http://localhost:11434/api/tags", timeout=10)
    if ollama_test.status_code == 200:
        print("   ✅ Ollama服务正常")
        print(f"   可用模型: {ollama_test.json()}")
    else:
        print(f"   ⚠️ Ollama响应异常: {ollama_test.status_code}")

    embeddings = OllamaEmbeddings(
        model="qwen2.5:3b",
        base_url="http://localhost:11434"
    )

    # 测试嵌入
    print("   测试嵌入模型...")
    test_text = "测试文本"
    test_vector = embeddings.embed_query(test_text)
    print(f"   ✅ 嵌入成功！向量维度: {len(test_vector)}")

except Exception as e:
    print(f"   ❌ 嵌入失败: {e}")
    print("   💡 请检查:")
    print("   1. Ollama是否运行: http://localhost:11434")
    print("   2. 模型是否下载: ollama pull qwen2.5:3b")
    exit()

# 4. 向量存储
try:
    print("\n4. 正在创建向量数据库...")
    start_time = time.time()
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory="./chroma_db"
    )
    end_time = time.time()
    print(f"   ✅ 创建成功！耗时: {end_time - start_time:.2f}秒")
    print(f"   向量库位置: ./chroma_db")
except Exception as e:
    print(f"   ❌ 向量数据库创建失败: {e}")
    exit()

# 5. LLM
try:
    print("\n5. 正在初始化LLM...")
    llm = ChatOllama(
        base_url="http://localhost:11434",
        model="qwen2.5:3b",
        temperature=0.1
    )

    # 测试LLM
    print("   测试LLM...")
    test_response = llm.invoke("Hello")
    print(f"   ✅ LLM正常！测试响应: {test_response.content[:50]}...")
except Exception as e:
    print(f"   ❌ LLM初始化失败: {e}")
    exit()

# 6. 创建问答链
try:
    print("\n6. 正在创建问答链...")
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=vectorstore.as_retriever(),
        return_source_documents=True
    )
    print("   ✅ 问答链创建成功！")
except Exception as e:
    print(f"   ❌ 问答链创建失败: {e}")
    exit()

print("\n" + "=" * 60)
print("🎉 RAG系统初始化完成！开始测试问答...")
print("=" * 60)

# 7. 测试问答
test_questions = [
    "这是关于什么的试卷？",
    "考试时间多久？",
    "听力部分有多少题？",
]

for i, question in enumerate(test_questions, 1):
    print(f"\n🔍 测试 {i}: {question}")
    try:
        start_time = time.time()
        result = qa_chain.invoke(question)
        end_time = time.time()

        print(f"   耗时: {end_time - start_time:.2f}秒")
        print(f"   回答: {result['result']}")

        # 显示来源
        if result.get('source_documents'):
            print(f"   参考了 {len(result['source_documents'])} 个文档片段")

    except Exception as e:
        print(f"   ❌ 问答失败: {e}")

print("\n" + "=" * 60)
print("✅ 测试完成！现在你可以输入自己的问题：")
print("输入 'quit' 退出")
print("=" * 60)

# 交互模式
while True:
    try:
        user_question = input("\n💭 请输入问题: ").strip()
        if user_question.lower() in ['quit', 'exit', 'q']:
            print("👋 再见！")
            break

        if not user_question:
            continue

        print("⏳ 正在生成回答...")
        result = qa_chain.invoke(user_question)
        print(f"\n💡 回答: {result['result']}")

    except KeyboardInterrupt:
        print("\n👋 用户中断，退出程序")
        break
    except Exception as e:
        print(f"❌ 出错: {e}")