老万】从0开始学chatGPT（十二）：如何让机器人痛说革命家史

class Memory:    """A memory stored in the PineCone index."""
    def __init__(self, id: str, time: datetime,                 importance: int, text: str) -> None:        self.id = id        self.time = time        self.importance = importance        self.text = text
    def __repr__(self) -> str:        return (f"Memory(id={self.id}, time={datetime_to_string(self.time)}, "                f"importance={self.importance}, text={self.text})")

def query_memory(    query: str,    top_k: int = 10) -> List[Tuple[float, Memory]]:    index = pinecone.Index(PINECONE_INDEX)    result = index.query(        namespace=PINECONE_INDEX_NAMESPACE,        vector=to_embedding(query),        top_k=top_k,        include_values=False,        include_metadata=True,    )    memories = []    for item in result["matches"]:        id = item["id"]        similarity = item["score"]        metadata = item["metadata"]        memory = Memory(            id=id,            time=string_to_datetime(id),            importance=metadata[PINECONE_INDEX_METADATA_KEY_MEMORY_IMPORTANCE],            text=metadata[PINECONE_INDEX_METADATA_KEY_MEMORY_TEXT],        )        memories.append((similarity, memory))    return memories

% src/main.py query "天气"2023-05-29T01:18:16.210647 (0.869009852) 今天天气不错,挺风和日丽的.2023-05-29T00:25:47.858912 (0.782007575) It's really sunny today.2023-05-29T01:19:43.154695 (0.751624584) 我在写一篇关于ChatGPT的文章.

% src/main.py query "GPT" 2023-05-29T01:19:43.154695 (0.79688251) 我在写一篇关于ChatGPT的文章.2023-05-29T00:25:47.858912 (0.722446561) It's really sunny today.2023-05-29T01:18:16.210647 (0.703824341) 今天天气不错,挺风和日丽的.

def comprehensive_score(        similarity: float, memory: Memory) -> float:    memory_age = datetime.now() - memory.time    recency_score = pow(0.99, memory_age.total_seconds() / 3600.0)    return similarity + recency_score + 0.3 * memory.importance / 10.0

def query_memory(    query: str,    start_time: str = "",    end_time: str = "",    top_k: int = 10,    scorer: Callable[[float, Memory], float] = comprehensive_score,) -> List[Tuple[float, Memory]]:    index = pinecone.Index(PINECONE_INDEX)    filter: Optional[Dict[str, Any]] = None    if start_time and end_time:        filter = {            "$and": [                {                    "time": {"$gte": string_to_timestamp_in_microseconds(start_time)},                },                {                    "time": {"$lt": string_to_timestamp_in_microseconds(end_time)},                },            ]        }    elif start_time:        filter = {"time": {"$gte": string_to_timestamp_in_microseconds(start_time)}}    elif end_time:        filter = {"time": {"$lt": string_to_timestamp_in_microseconds(end_time)}}
    result = index.query(        namespace=PINECONE_INDEX_NAMESPACE,        vector=to_embedding(query),        filter=filter,        top_k=top_k,        include_values=False,        include_metadata=True,    )    memories = []    for item in result["matches"]:        id = item["id"]        similarity = item["score"]        metadata = item["metadata"]        memory = Memory(            id=id,            time=string_to_datetime(id),            importance=metadata[PINECONE_INDEX_METADATA_KEY_MEMORY_IMPORTANCE],            text=metadata[PINECONE_INDEX_METADATA_KEY_MEMORY_TEXT],        )        memories.append(            (                scorer(similarity, memory),                memory,            )        )
    return sorted(memories, key=lambda x: x[0], reverse=True)

def chat() -> None:    print("Type 'quit' to exit.", file=sys.stderr)    conversation: List[str] = []    while True:        text = input(">>> ").strip()        if not text:            continue        if text == "quit":            break
        end_time = datetime.now() - timedelta(hours=1)        top_memories = query_memory(            query=text,            end_time=datetime_to_string(end_time))[:3]
        conversation.append(            _make_prompt(text, top_memories))        add_memory(f"I said: ```{text}```")
        answer = _get_gpt_answer(            _make_messages(conversation))        print(f"{answer}", file=sys.stderr)
        conversation.append(answer)        add_memory(f"GPT said: ```{answer}```")

[ "如果没有外界帮助，你能记住多少上下文？",  "我可以记住大约几千个令牌的上下文，...",  "什么是向量数据库？",  "向量数据库是一种专门用于处理向量数据的数据库。...",]

def _make_prompt(text: str,                 top_memories: List[Tuple[float, Memory]]) -> str:    prompt = "Here are some memories (separated by !!!! and ended by $$$$) that might be relevant:\n"    for _, memory in top_memories:        prompt += f"""!!!!At {memory.time}, {memory.text}"""    prompt += f"""$$$$With that...{text}"""    return prompt

def _make_messages(        conversation: List[str]) -> List[Dict[str, str]]:    assert (        len(conversation) % 2 == 1    ), "The conversation history should have an odd length."
    reverse_messages: List[Dict[str, str]] = []    role = "user"  # user or assistant.    total_len = 0    for text in reversed(conversation):        # Limit the total length of text sent to the bot.        total_len += len(text)        if total_len > 2048:            break
        # In Python, it's more efficient to append to a list than to insert at the beginning.        reverse_messages.append({"role": role, "content": text})        # Reverse the role.        role = "assistant" if role == "user" else "user"
    # Set the behavior of the bot.    reverse_messages.append(        {"role": "system", "content": "You are a helpful assistant."}    )    return list(reversed(reverse_messages))

def _get_gpt_answer(        messages: List[Dict[str, str]],        temperature: float = 0.7) -> str:    response = openai.ChatCompletion.create(        model="gpt-3.5-turbo",        messages=messages,        temperature=temperature,    )    return response["choices"][0]["message"]["content"].strip()