Prompt编排与会话历史#

1
# pip install langchain langchain-openai
2

3
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
4

5
prompt = ChatPromptTemplate.from_messages([
6
    ("system", "你是一位技术支持助手。"),
7
    MessagesPlaceholder(variable_name="history"),
8
    ("human", "{input}"),
9
])

1
from langchain_core.messages import HumanMessage, AIMessage
2

3
prompt_value = prompt.invoke({
4
    "history": [
5
        HumanMessage(content="我在部署时遇到报错。"),
6
        AIMessage(content="请把报错信息贴出来。"),
7
    ],
8
    "input": "报错和数据库连接有关。",
9
})

1
# pip install langchain
2

3
from langchain_core.prompts import ChatPromptTemplate
4

5
prompt = ChatPromptTemplate.from_messages([
6
    ("system", "你是 {company} 的技术文档助手，请始终使用 {language} 回答。"),
7
    ("human", "{question}"),
8
]).partial(
9
    company="OpenDocs",
10
    language="中文",
11
)
12

13
messages = prompt.invoke({"question": "什么是向量数据库？"})

1
# pip install langchain langchain-openai langchain-community faiss-cpu
2

3
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
4
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
5
from langchain_openai import OpenAIEmbeddings
6
from langchain_community.vectorstores import FAISS
7

8
examples = [
9
    {"question": "什么是 Python 装饰器？", "answer": "装饰器是用于包装函数的高阶函数。"},
10
    {"question": "什么是事务隔离级别？", "answer": "事务隔离级别定义并发事务之间的可见性。"},
11
    {"question": "什么是向量数据库？", "answer": "向量数据库用于存储和检索高维向量。"},
12
]
13

14
selector = SemanticSimilarityExampleSelector.from_examples(
15
    examples=examples,
16
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
17
    vectorstore_cls=FAISS,
18
    k=2,
19
    input_keys=["question"],
20
)
21

22
example_prompt = ChatPromptTemplate.from_messages([
23
    ("human", "{question}"),
24
    ("ai", "{answer}"),
25
])
26

27
few_shot_prompt = FewShotChatMessagePromptTemplate(
28
    example_selector=selector,
29
    example_prompt=example_prompt,
30
)
31

32
final_prompt = ChatPromptTemplate.from_messages([
33
    ("system", "你是一位资深技术讲师，请用简洁准确的方式回答。"),
34
    few_shot_prompt,
35
    ("human", "{question}"),
36
])

1
# pip install langchain langchain-openai
2

3
from langchain_openai import ChatOpenAI
4
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
5
from langchain_core.chat_history import InMemoryChatMessageHistory
6
from langchain_core.runnables.history import RunnableWithMessageHistory
7

8
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
9

10
prompt = ChatPromptTemplate.from_messages([
11
    ("system", "你是一个会记住上下文的助手。"),
12
    MessagesPlaceholder(variable_name="history"),
13
    ("human", "{input}"),
14
])
15

16
chain = prompt | llm
17
store = {}
18

19
def get_session_history(session_id: str):
20
    if session_id not in store:
21
        store[session_id] = InMemoryChatMessageHistory()
22
    return store[session_id]
23

24
chain_with_history = RunnableWithMessageHistory(
25
    chain,
26
    get_session_history,
27
    input_messages_key="input",
28
    history_messages_key="history",
29
)
30

31
config = {"configurable": {"session_id": "user_001"}}
32

33
resp1 = chain_with_history.invoke({"input": "我叫小明。"}, config=config)
34
resp2 = chain_with_history.invoke({"input": "我叫什么名字？"}, config=config)

1
# pip install langchain-community sqlalchemy
2

3
from langchain_core.messages import HumanMessage
4
from langchain_community.chat_message_histories import SQLChatMessageHistory
5

6
history = SQLChatMessageHistory(
7
    session_id="user_001",
8
    connection="sqlite:///chat_history.db",
9
)
10

11
history.add_messages([
12
    HumanMessage(content="你好，我想了解 LangChain。"),
13
])
14

15
print(history.messages)

1
# pip install langchain-community redis
2

3
from langchain_community.chat_message_histories import RedisChatMessageHistory
4

5
history = RedisChatMessageHistory(
6
    session_id="user_001",
7
    url="redis://localhost:6379/0",
8
    ttl=3600,
9
)