Dynamodb 聊天消息历史

This notebook goes over how to use Dynamodb to store chat message history.

首先确保您已正确配置 AWS CLI。然后确保您已安装 boto3。

接下来，创建我们将存储消息的 DynamoDB 表：

import boto3

# 获取服务资源。
dynamodb = boto3.resource("dynamodb")

# 创建 DynamoDB 表。
table = dynamodb.create_table(
    TableName="SessionTable",
    KeySchema=[{"AttributeName": "SessionId", "KeyType": "HASH"}],
    AttributeDefinitions=[{"AttributeName": "SessionId", "AttributeType": "S"}],
    BillingMode="PAY_PER_REQUEST",
)

# 等待表存在。
table.meta.client.get_waiter("table_exists").wait(TableName="SessionTable")

# 打印一些关于表的数据。
print(table.item_count)

    0

DynamoDBChatMessageHistory (DynamoDB 聊天消息历史)

from langchain.memory.chat_message_histories import DynamoDBChatMessageHistory

history = DynamoDBChatMessageHistory(table_name="SessionTable", session_id="0")

history.add_user_message("hi!")

history.add_ai_message("whats up?")

history.messages

    [HumanMessage(content='hi!', additional_kwargs={}, example=False),
     AIMessage(content='whats up?', additional_kwargs={}, example=False)]

使用自定义端点 URL 的 DynamoDBChatMessageHistory (使用自定义端点 URL 的 DynamoDB 聊天消息历史)

有时候指定连接到 AWS 端点的 URL 是有用的。例如，当您在本地运行 Localstack 时。对于这些情况，您可以通过构造函数中的 endpoint_url 参数指定 URL。

from langchain.memory.chat_message_histories import DynamoDBChatMessageHistory

history = DynamoDBChatMessageHistory(
    table_name="SessionTable",
    session_id="0",
    endpoint_url="http://localhost.localstack.cloud:4566",
)

使用 DynamoDB 内存的 Agent (使用 DynamoDB 内存的 Agent)

from langchain.agents import Tool
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent
from langchain.agents import AgentType
from langchain.utilities import PythonREPL
from getpass import getpass

message_history = DynamoDBChatMessageHistory(table_name="SessionTable", session_id="1")
memory = ConversationBufferMemory(
    memory_key="chat_history", chat_memory=message_history, return_messages=True
)

python_repl = PythonREPL()

# 您可以创建要传递给代理的工具
tools = [
    Tool(
        name="python_repl",
        description="A Python shell. Use this to execute python commands. Input should be a valid python command. If you want to see the output of a value, you should print it out with `print(...)`.",
        func=python_repl.run,
    )
]

llm = ChatOpenAI(temperature=0)
agent_chain = initialize_agent(
    tools,
    llm,
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    verbose=True,
    memory=memory,
)

agent_chain.run(input="Hello!")

    
    
    > Entering new AgentExecutor chain...
    {
        "action": "Final Answer",
        "action_input": "Hello! How can I assist you today?"
    }
    
    > Finished chain.





    'Hello! How can I assist you today?'

agent_chain.run(input="Who owns Twitter?")

    
    
    > Entering new AgentExecutor chain...
    {
        "action": "python_repl",
        "action_input": "import requests\nfrom bs4 import BeautifulSoup\n\nurl = 'https://en.wikipedia.org/wiki/Twitter'\nresponse = requests.get(url)\nsoup = BeautifulSoup(response.content, 'html.parser')\nowner = soup.find('th', text='Owner').find_next_sibling('td').text.strip()\nprint(owner)"
    }
    Observation: X Corp. (2023–present)Twitter, Inc. (2006–2023)
    
    Thought:{
        "action": "Final Answer",
        "action_input": "X Corp. (2023–present)Twitter, Inc. (2006–2023)"
    }
    
    > Finished chain.





    'X Corp. (2023–present)Twitter, Inc. (2006–2023)'

agent_chain.run(input="My name is Bob.")

    
    
    > Entering new AgentExecutor chain...
    {
        "action": "Final Answer",
        "action_input": "Hello Bob! How can I assist you today?"
    }
    
    > Finished chain.





    'Hello Bob! How can I assist you today?'

agent_chain.run(input="Who am I?")

    
    
    > Entering new AgentExecutor chain...
    {
        "action": "Final Answer",
        "action_input": "Your name is Bob."
    }
    
    > Finished chain.





    'Your name is Bob.'