1
from langchain_core.prompts import ChatPromptTemplate
2

3
# 方式一：from_messages（推荐）
4
prompt = ChatPromptTemplate.from_messages([
5
    ("system", "你是一个{role}，专门帮助用户解决{topic}问题"),
6
    ("human", "我的问题是：{question}")
7
])
8

9
# 方式二：from_template（传统方式）
10
prompt = ChatPromptTemplate.from_template(
11
    "你是{role}，请用{style}风格回答以下问题：{question}"
12
)
13

14
# 格式化输入
15
formatted_prompt = prompt.format(
16
    role="技术顾问",
17
    topic="编程",
18
    question="Python如何处理异常？",
19
    style="简洁专业"
20
)
21

22
print(formatted_prompt)

1
System: 你是一个技术顾问，专门帮助用户解决编程问题
2
Human: 我的问题是：Python如何处理异常？

1
from langchain_openai import ChatOpenAI
2
from langchain_anthropic import ChatAnthropic
3
from langchain_google_genai import ChatGoogleGenerativeAI
4

5
# OpenAI
6
openai_llm = ChatOpenAI(
7
    model="gpt-4o",
8
    temperature=0.7,
9
    streaming=True  # 启用流式输出
10
)
11

12
# Anthropic
13
claude_llm = ChatAnthropic(
14
    model="claude-3-5-sonnet-20241022",
15
    temperature=0.7
16
)
17

18
# Google Gemini
19
gemini_llm = ChatGoogleGenerativeAI(
20
    model="gemini-2.0-flash-exp",
21
    temperature=0.7
22
)

1
from langchain_core.output_parsers import StrOutputParser
2
from langchain_core.output_parsers import JsonOutputParser
3
from langchain.output_parsers import PydanticOutputParser
4
from pydantic import BaseModel
5

6
# 字符串解析器（最常用）
7
str_parser = StrOutputParser()
8

9
# JSON解析器
10
json_parser = JsonOutputParser()
11

12
# Pydantic解析器（推荐用于结构化输出）
13
class Answer(BaseModel):
14
    result: str
15
    confidence: float
16
    sources: list[str]
17

18
pydantic_parser = PydanticOutputParser(pydantic_object=Answer)
19

20
# 使用示例
21
response = llm.invoke("What is 2+2?")
22
parsed = str_parser.invoke(response)
23
print(parsed)  # "2"

1
from langchain_openai import ChatOpenAI
2
from langchain_core.prompts import ChatPromptTemplate
3
from langchain_core.output_parsers import StrOutputParser
4

5
# 初始化模型
6
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
7

8
# 创建Prompt模板
9
prompt = ChatPromptTemplate.from_template(
10
    "用一句话解释{concept}，用中文回答"
11
)
12

13
# 创建输出解析器
14
parser = StrOutputParser()
15

16
# 方式一：传统方式（手动调用）
17
formatted_prompt = prompt.format(concept="LLM")
18
response = llm.invoke(formatted_prompt)
19
result = parser.invoke(response)
20
print(result)
21

22
# 方式二：LCEL管道方式（推荐）
23
chain = prompt | llm | parser
24
result = chain.invoke({"concept": "LLM"})
25
print(result)

1
from langchain_openai import ChatOpenAI
2
from langchain_core.prompts import ChatPromptTemplate
3
from langchain.chains import LLMChain
4

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

7
prompt = ChatPromptTemplate.from_template(
8
    "帮我写一首关于{topic}的诗"
9
)
10

11
chain = LLMChain(llm=llm, prompt=prompt)
12
result = chain.run(topic="春天")
13
print(result)

1
from langchain.chains import SequentialChain
2
from langchain.chains import LLMChain
3
from langchain_core.prompts import ChatPromptTemplate
4

5
# 链1：生成标题
6
title_chain = LLMChain(
7
    llm=llm,
8
    prompt=ChatPromptTemplate.from_template("为以下内容生成一个标题：{content}"),
9
    output_key="title"
10
)
11

12
# 链2：生成摘要
13
summary_chain = LLMChain(
14
    llm=llm,
15
    prompt=ChatPromptTemplate.from_template("用50字概括以下内容：{content}"),
16
    output_key="summary"
17
)
18

19
# 组合顺序链
20
sequential_chain = SequentialChain(
21
    chains=[title_chain, summary_chain],
22
    input_variables=["content"],
23
    output_variables=["title", "summary"]
24
)
25

26
result = sequential_chain.invoke({"content": "Python是一门易学难精的编程语言..."})
27
print(result)

1
from langchain_core.runnables import Runnable
2

3
# Runnable接口的核心方法
4
class Runnable:
5
    def invoke(self, input, config=None):
6
        """同步调用"""
7
        pass
8

9
    async def ainvoke(self, input, config=None):
10
        """异步调用"""
11
        pass
12

13
    def stream(self, input, config=None):
14
        """流式输出"""
15
        pass
16

17
    def batch(self, inputs, config=None):
18
        """批量处理"""
19
        pass

1
# 基本语法
2
chain = component_a | component_b | component_c
3

4
# 等价于
5
result = component_c.invoke(component_b.invoke(component_a.invoke(input)))

1
from langchain_openai import ChatOpenAI
2
from langchain_core.prompts import ChatPromptTemplate
3
from langchain_core.output_parsers import StrOutputParser
4

5
llm = ChatOpenAI(model="gpt-4o-mini")
6
prompt = ChatPromptTemplate.from_template("用中文介绍{topic}")
7
parser = StrOutputParser()
8

9
# 手动串联
10
formatted = prompt.format(topic="人工智能")
11
response = llm.invoke(formatted)
12
result = parser.invoke(response)
13
print(result)

1
# 一行搞定
2
chain = prompt | llm | parser
3
result = chain.invoke({"topic": "人工智能"})
4
print(result)

1
from langchain_core.runnables import RunnableParallel, RunnableBranch
2

3
# 并行执行多个分支
4
parallel_chain = RunnableParallel(
5
    chinese = prompt | llm | parser,
6
    english = (prompt | llm | parser).bind(language="en"),
7
    code = code_prompt | llm | parser
8
)
9

10
# 条件分支
11
branch_chain = RunnableBranch(
12
    (lambda x: x.get("type") == "code", code_chain),
13
    (lambda x: x.get("type") == "data", data_chain),
14
    default_chain  # 默认链
15
)

1
chain = prompt | llm | parser
2

3
# 流式输出（一个字一个字显示）
4
for chunk in chain.stream({"topic": "量子计算"}):
5
    print(chunk, end="", flush=True)

1
from langchain.memory import ConversationBufferMemory
2
from langchain.chains import LLMChain
3
from langchain_openai import ChatOpenAI
4
from langchain_core.prompts import ChatPromptTemplate
5

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

8
# 创建带记忆的链
9
memory = ConversationBufferMemory(
10
    memory_key="history",
11
    return_messages=True  # 返回消息对象列表
12
)
13

14
prompt = ChatPromptTemplate.from_template(
15
    """基于以下对话历史回答问题：
16

17
历史：
18
{history}
19

20
问题：{question}"""
21
)
22

23
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)
24

25
# 多轮对话
26
print(chain.invoke({"question": "我叫张三"}))
27
print(chain.invoke({"question": "我叫什么名字？"}))

1
from langchain.memory import ConversationBufferWindowMemory
2

3
memory = ConversationBufferWindowMemory(
4
    k=3,  # 只保留最近3轮对话
5
    memory_key="chat_history",
6
    return_messages=True
7
)

1
from langchain.memory import ConversationSummaryMemory
2

3
memory = ConversationSummaryMemory(
4
    llm=llm,  # 用于生成摘要的LLM
5
    memory_key="summary",
6
    return_messages=True
7
)

1
import faiss
2
from langchain.docstore import InMemoryDocstore
3
from langchain_community.vectorstores import FAISS
4
from langchain_openai import OpenAIEmbeddings
5
from langchain.memory import VectorStoreRetrieverMemory
6
from langchain.chains import ConversationChain
7
from langchain_openai import ChatOpenAI
8
from langchain_core.prompts import ChatPromptTemplate
9

10
# 初始化嵌入模型和LLM
11
embedding = OpenAIEmbeddings()
12
llm = ChatOpenAI(model="gpt-4o-mini")
13

14
# 创建FAISS向量存储
15
embedding_size = 1536  # OpenAIEmbeddings维度
16
index = faiss.IndexFlatL2(embedding_size)
17
vectorstore = FAISS(index, embedding, InMemoryDocStore(), {})
18

19
# 创建retriever（k=3表示返回最相关的3条记忆）
20
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
21

22
# 创建向量记忆
23
memory = VectorStoreRetrieverMemory(
24
    retriever=retriever,
25
    memory_key="chat_history"
26
)
27

28
# 保存一些对话记忆
29
memory.save_context(
30
    {"input": "我最喜欢的食物是披萨"},
31
    {"output": "好的，我记住了 你喜欢披萨"}
32
)
33
memory.save_context(
34
    {"input": "我喜欢在周末去跑步"},
35
    {"output": "运动是个很好的习惯！"}
36
)
37
memory.save_context(
38
    {"input": "我最近在学习Python编程"},
39
    {"output": "Python是一门很实用的语言，继续加油！"}
40
)
41

42
# 检索相关记忆
43
print(memory.load_memory_variables({"prompt": "我应该吃什么？"}))

1
{'chat_history': 'input: 我最喜欢的食物是披萨\noutput: 好的，我记住了...'}

1
from langchain_community.vectorstores import Chroma
2
from langchain.embeddings import OpenAIEmbeddings
3

4
# 创建Chroma向量存储
5
vectorstore = Chroma.from_documents(
6
    documents=[],  # 初始为空
7
    embedding=OpenAIEmbeddings()
8
)
9

10
# 创建memory
11
memory = VectorStoreRetrieverMemory(
12
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
13
    memory_key="history"
14
)

1
from langchain_pinecone import PineconeVectorStore
2
from langchain.embeddings import OpenAIEmbeddings
3
from langchain.memory import VectorStoreRetrieverMemory
4

5
# 连接Pinecone
6
vectorstore = PineconeVectorStore.from_params(
7
    index_name="my-index",
8
    embedding=OpenAIEmbeddings(),
9
    namespace="memory"
10
)
11

12
memory = VectorStoreRetrieverMemory(
13
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
14
    memory_key="long_term_memory"
15
)

1
# 创建对话链
2
prompt = ChatPromptTemplate.from_template("""
3
基于以下记忆回答用户问题。如果没有相关信息，请如实说明。
4

5
记忆：
6
{chat_history}
7

8
问题：{input}
9
""")
10

11
conversation = ConversationChain(
12
    llm=llm,
13
    memory=memory,
14
    prompt=prompt,
15
    verbose=True
16
)
17

18
# 第一次对话
19
response1 = conversation.invoke({"input": "我的爱好是什么？"})
20
print(response1)
21

22
# 第二次对话（跨会话）
23
response2 = conversation.invoke({"input": "我应该学习什么编程语言？"})
24
print(response2)

1
retriever = vectorstore.as_retriever(
2
    search_type="similarity",  # 默认
3
    search_kwargs={"k": 3}
4
)

1
retriever = vectorstore.as_retriever(
2
    search_type="mmr",  # 避免返回相似度过高的结果
3
    search_kwargs={"k": 5, "fetch_k": 20}
4
)

1
retriever = vectorstore.as_retriever(
2
    search_kwargs={"k": 5, "score_threshold": 0.7}
3
)

1
from langgraph.checkpoint.memory import MemorySaver
2
from langgraph.graph import StateGraph, START
3

4
# 创建带持久化的图
5
checkpointer = MemorySaver()
6

7
graph = StateGraph(GraphState)
8
graph.add_node("process", process_node)
9
graph.add_edge(START, "process")
10
graph.compile(checkpointer=checkpointer)
11

12
# 线程ID用于恢复状态
13
config = {"configurable": {"thread_id": "user_123"}}
14
result = graph.invoke({"input": "hello"}, config)
15

16
# 恢复状态继续执行
17
result = graph.invoke({"input": "continue"}, config)

1
from langgraph.graph import StateGraph, START, END
2

3
# 1. 定义状态类型
4
class GraphState(TypedDict):
5
    input: str
6
    result: str
7

8
# 2. 定义节点函数
9
def node_a(state: GraphState) -> GraphState:
10
    return {"result": f"处理: {state['input']}"}
11

12
def node_b(state: GraphState) -> GraphState:
13
    return {"result": state["result"] + " + 节点B"}
14

15
# 3. 构建图
16
graph = StateGraph(GraphState)
17
graph.add_node("A", node_a)
18
graph.add_node("B", node_b)
19
graph.add_edge(START, "A")    # 起点 → A
20
graph.add_edge("A", "B")      # A → B
21
graph.add_edge("B", END)      # B → 终点
22

23
# 4. 编译并执行
24
app = graph.compile()
25
result = app.invoke({"input": "hello"})
26
print(result)  # {'input': 'hello', 'result': '处理: hello + 节点B'}