構建RAG智能體(2)：運行狀態鏈

在現代AI應用開發中，如何讓聊天機器人具備記憶能力和上下文理解是一個核心挑戰。傳統的無狀態對話系統往往無法處理復雜的多輪對話場景，特別是當用戶需要提供多種信息來完成特定任務時。

本文就來討論一下如何利用runnable來編排更有趣的語言模型系統，并理解如何使用運行狀態鏈來管理復雜的對話策略和執行長篇文檔的推理。

文章目錄

1 保持變量流動
2 運行狀態鏈
3 使用運行狀態鏈實現知識庫
4 航空公司客服機器人
5 總結

1 保持變量流動

在之前的示例中，我們通過創建、改變和消費狀態，在獨立的鏈中實現了有趣的邏輯。這些狀態以帶有描述性鍵和有用值的字典形式傳遞，這些值被用來為后續的程序提供它們操作所需的信息。回憶一下上一篇文章中的零樣本分類示例：

%%time
## ^^ This notebook is timed, which will print out how long it all took
from langchain_core.runnables import RunnableLambda
from langchain_nvidia_ai_endpoints import ChatNVIDIA
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from typing import List, Union
from operator import itemgetter## Zero-shot classification prompt and chain w/ explicit few-shot prompting
sys_msg = ("Choose the most likely topic classification given the sentence as context."" Only one word, no explanation.\n[Options : {options}]"
)zsc_prompt = ChatPromptTemplate.from_template(f"{sys_msg}\n\n""[[The sea is awesome]][/INST]boat</s><s>[INST]""[[{input}]]"
)## Define your simple instruct_model
instruct_chat = ChatNVIDIA(model="mistralai/mistral-7b-instruct-v0.2")
instruct_llm = instruct_chat | StrOutputParser()
one_word_llm = instruct_chat.bind(stop=[" ", "\n"]) | StrOutputParser()zsc_chain = zsc_prompt | one_word_llm## Function that just prints out the first word of the output. With early stopping bind
def zsc_call(input, options=["car", "boat", "airplane", "bike"]):return zsc_chain.invoke({"input" : input, "options" : options}).split()[0]print("-" * 80)
print(zsc_call("Should I take the next exit, or keep going to the next one?"))print("-" * 80)
print(zsc_call("I get seasick, so I think I'll pass on the trip"))print("-" * 80)
print(zsc_call("I'm scared of heights, so flying probably isn't for me"))

輸出：

--------------------------------------------------------------------------------
car
--------------------------------------------------------------------------------
boat
--------------------------------------------------------------------------------
air
CPU times: user 23.4 ms, sys: 12.9 ms, total: 36.3 ms
Wall time: 1.29 s

這個鏈做出的幾個設計決策使其非常易于使用，其中最關鍵的一點是：我們希望它像一個函數一樣運作，所以我們只希望它生成輸出并返回它。

這使得該鏈非常自然地可以作為一個模塊包含在更大的鏈系統中。例如，下面的鏈將接受一個字符串，提取最可能的主題，然后根據該主題生成一個新句子：

%%time
## ^^ 這個筆記被計時，會打印出總共花費的時間
gen_prompt = ChatPromptTemplate.from_template("Make a new sentence about the the following topic: {topic}. Be creative!"
)gen_chain = gen_prompt | instruct_llminput_msg = "I get seasick, so I think I'll pass on the trip"
options = ["car", "boat", "airplane", "bike"]chain = (## -> {"input", "options"}{'topic' : zsc_chain}| PPrint()## -> {**, "topic"}| gen_chain## -> string
)chain.invoke({"input" : input_msg, "options" : options})

輸出：

State:
{'topic': 'boat'}CPU times: user 23.6 ms, sys: 4.28 ms, total: 27.8 ms
Wall time: 1.03 s" As the sun began to set, the children's eyes gleamed with excitement as they rowed their makeshift paper boat through the sea of rippling bathwater in the living room, creating waves of laughter and magic."

然而，當您想保持信息流動時，這會有點問題，因為我們在生成響應時丟失了主題和輸入變量。在簡單的鏈式結構中，只傳遞上一步的輸出，導致你無法同時訪問input和topic。如果你想讓模型同時參考兩個變量，需要用運行狀態鏈或狀態管理的方式把它們都傳過去。如果我們想同時使用輸出和輸入做些什么，我們需要一種方法來確保兩個變量都能傳遞過去。

我們可以使用映射runnable(即從字典解釋或使用手動的RunnableMap)來將兩個變量都傳遞過去，方法是將我們鏈的輸出分配給一個單一的鍵，并讓其他鍵按需傳播。或者，我們也可以使用RunnableAssign來默認將消耗狀態的鏈的輸出與輸入字典合并。

通過這種方式，我們可以在我們的鏈系統中傳播任何我們想要的東西：

%%time
## ^^ 這個筆記被計時，會打印出總共花費的時間from langchain.schema.runnable import RunnableBranch, RunnablePassthrough
from langchain.schema.runnable.passthrough import RunnableAssign
from functools import partialbig_chain = (PPrint()## 手動映射。有時在分支鏈內部很有用| {'input' : lambda d: d.get('input'), 'topic' : zsc_chain}| PPrint()## RunnableAssign傳遞。默認情況下更適合運行狀態鏈| RunnableAssign({'generation' : gen_chain})| PPrint()## 一起使用輸入和生成的內容| RunnableAssign({'combination' : (ChatPromptTemplate.from_template("Consider the following passages:""\nP1: {input}""\nP2: {generation}""\n\nCombine the ideas from both sentences into one simple one.")| instruct_llm)})
)output = big_chain.invoke({"input" : "I get seasick, so I think I'll pass on the trip","options" : ["car", "boat", "airplane", "bike", "unknown"]
})
pprint("Final Output: ", output)

輸出：

State: 
{'input': "I get seasick, so I think I'll pass on the trip",'options': ['car', 'boat', 'airplane', 'bike', 'unknown']
}
State: 
{'input': "I get seasick, so I think I'll pass on the trip", 'topic': ' boat'}
State: 
{'input': "I get seasick, so I think I'll pass on the trip",'topic': ' boat','generation': " As the sun began to set, the children's eyes gleamed with excitement as they rowed their 
makeshift paper boat through the sea of rippling bathwater in the living room, creating waves of laughter and 
magic."
}
Final Output: 
{'input': "I get seasick, so I think I'll pass on the trip",'topic': ' boat','generation': " As the sun began to set, the children's eyes gleamed with excitement as they rowed their 
makeshift paper boat through the sea of rippling bathwater in the living room, creating waves of laughter and 
magic.",'combination': "Feeling seasick, I'll have to sit this one out, as the children excitedly navigate their paper 
boat in the makeshift ocean of our living room, their laughter filling the air."
}

2 運行狀態鏈

上面只是一個簡單例子，如果說有什么作用的話，那就是展示了將許多LLM調用鏈接在一起進行內部推理的缺點。然而，保持信息在鏈中流動對于制作能夠累積有用狀態信息或以多遍方式操作的復雜鏈來說是無價的。

具體來說，一個非常簡單但有效的鏈是運行狀態鏈，它強制執行以下屬性：

**“運行狀態”**是一個字典，包含系統關心的所有變量。
**“分支”**是一個可以引入運行狀態并可以將其降級為響應的鏈。
分支只能在RunnableAssign作用域內運行，并且分支的輸入應來自運行狀態。

在這里插入圖片描述

你可以將運行狀態鏈抽象看作是帶有狀態變量(或屬性)和函數(或方法)的Pythonic類的函數式變體。

鏈就像是包裝所有功能的抽象類。
運行狀態就像是屬性(應該總是可訪問的)。
分支就像是類方法(可以選擇使用哪些屬性)。
.invoke 或類似的過程就像是按順序運行分支的__call__方法。

通過在鏈中強制執行這種范式：

可以保持狀態變量在您的鏈中傳播，允許內部組件訪問任何必要的東西，并為以后使用累積狀態值。
還可以將您的鏈的輸出作為輸入傳回，允許一個“while-循環”式的鏈，不斷更新和構建您的運行狀態。

3 使用運行狀態鏈實現知識庫

在理解了運行狀態鏈的基本結構和原理之后，我們可以探索如何將這種方法擴展到管理更復雜的任務，特別是在創建通過交互演變的動態系統中。本節將重點介紹如何實現一個使用json啟用的槽位填充累積的知識庫：

知識庫(Knowledge Base)：一個信息存儲庫，用于讓我們的LLM跟蹤相關信息。
JSON啟用的槽位填充(JSON-Enabled Slot Filling)：要求一個經過指令調優的模型輸出一個json風格的格式(可以包括一個字典)，其中包含一系列槽位，并依賴LLM用有用和相關的信息來填充這些槽位。

定義我們的知識庫

為了構建一個響應迅速且智能的系統，我們需要一種方法，不僅能處理輸入，還能在對話流程中保留和更新基本信息。這就是LangChain和Pydantic結合的關鍵所在。

**Pydantic**是一個流行的Python驗證庫，在構建和驗證數據模型方面起著重要作用。作為其特性之一，Pydantic提供了結構化的模型類，用簡化的語法和深度的定制選項來驗證對象(數據、類、它們自身等)。這個框架在整個LangChain中被廣泛使用，并且在涉及數據轉換的用例中成為一個必要的組件。

我們可以先構建一個BaseModel類并定義一些Field變量來創建一個結構化的知識庫，如下所示：

from pydantic import BaseModel, Field
from typing import Dict, Union, Optionalinstruct_chat = ChatNVIDIA(model="mistralai/mistral-7b-instruct-v0.2")class KnowledgeBase(BaseModel):## BaseModel的字段，當知識庫被構建時將被驗證/分配topic: str = Field('general', description="Current conversation topic")user_preferences: Dict[str, Union[str, int]] = Field({}, description="User preferences and choices")session_notes: list = Field([], description="Notes on the ongoing session")unresolved_queries: list = Field([], description="Unresolved user queries")action_items: list = Field([], description="Actionable items identified during the conversation")print(repr(KnowledgeBase(topic = "Travel")))
輸出:
KnowledgeBase(topic='Travel', user_preferences={}, session_notes=[], unresolved_queries=[], action_items=[])

這種方法的真正優勢在于LangChain提供的額外的以LLM為中心的功能，我們可以將其集成到我們的用例中。其中一個特性是PydanticOutputParser，它增強了Pydantic對象的能力，比如自動生成格式說明。

from langchain.output_parsers import PydanticOutputParserinstruct_string = PydanticOutputParser(pydantic_object=KnowledgeBase).get_format_instructions()
pprint(instruct_string)

輸出：

The output should be formatted as a JSON instance that conforms to the JSON schema below.As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": 
"array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": 
["bar", "baz"]}} is not well-formatted.Here is the output schema:
```
{"properties": {"topic": {"default": "general", "description": "Current conversation topic", "title": "Topic", 
"type": "string"}, "user_preferences": {"additionalProperties": {"anyOf": [{"type": "string"}, {"type": 
"integer"}]}, "default": {}, "description": "User preferences and choices", "title": "User Preferences", "type": 
"object"}, "session_notes": {"default": [], "description": "Notes on the ongoing session", "items": {}, "title": 
"Session Notes", "type": "array"}, "unresolved_queries": {"default": [], "description": "Unresolved user queries", 
"items": {}, "title": "Unresolved Queries", "type": "array"}, "action_items": {"default": [], "description": 
"Actionable items identified during the conversation", "items": {}, "title": "Action Items", "type": "array"}}}

這個功能為創建知識庫的有效輸入生成了指令，這反過來又通過提供一個具體的、期望輸出格式的單樣本示例來幫助LLM。

可運行的提取模塊(Runnable Extraction Module)

我們可以創建一個Runnable，它包裝了我們Pydantic類的功能，并簡化了知識庫的提示、生成和更新過程：

## RExtract的定義
def RExtract(pydantic_class, llm, prompt):'''Runnable提取模塊返回一個通過槽位填充提取來填充的知識字典'''parser = PydanticOutputParser(pydantic_object=pydantic_class)instruct_merge = RunnableAssign({'format_instructions' : lambda x: parser.get_format_instructions()})def preparse(string):if '{' not in string: string = '{' + stringif '}' not in string: string = string + '}'string = (string.replace("\\_", "_").replace("\n", " ").replace("\]", "]").replace("\[", "["))# print(string)  ## 適合用于診斷return stringreturn instruct_merge | prompt | llm | preparse | parser## RExtract的實際使用parser_prompt = ChatPromptTemplate.from_template("Update the knowledge base: {format_instructions}. Only use information from the input.""\n\nNEW MESSAGE: {input}"
)extractor = RExtract(KnowledgeBase, instruct_llm, parser_prompt)knowledge = extractor.invoke({'input' : "I love flowers so much! The orchids are amazing! Can you buy me some?"})
pprint(knowledge)

注意，由于LLM預測的模糊性，這個過程可能會失敗，特別是對于那些沒有針對指令跟隨進行優化的模型(還沒有被訓練成能理解并執行自然語言指令的助手型模型)。對于這個過程，擁有一個強大的指令跟隨LLM以及額外的檢查和失敗處理程序非常重要。

KnowledgeBase(topic='Flowers and orchids',user_preferences={'orchids': 'flowers I love'},session_notes=[],unresolved_queries=[],action_items=[]
)

動態知識庫更新

最后，我們可以創建一個在整個對話過程中不斷更新知識庫的系統。這是通過將知識庫的當前狀態連同新的用戶輸入一起反饋到系統中進行持續更新來完成的。

以下是一個示例系統，它既展示了該公式在填充細節方面的強大能力，也展示了假設填充性能會和一般響應性能一樣好的局限性：

class KnowledgeBase(BaseModel):firstname: str = Field('unknown', description="Chatting user's first name, unknown if unknown")lastname: str = Field('unknown', description="Chatting user's last name, unknown if unknown")location: str = Field('unknown', description="Where the user is located")summary: str = Field('unknown', description="Running summary of conversation. Update this with new input")response: str = Field('unknown', description="An ideal response to the user based on their new message")parser_prompt = ChatPromptTemplate.from_template("You are chatting with a user. The user just responded ('input'). Please update the knowledge base."" Record your response in the 'response' tag to continue the conversation."" Do not hallucinate any details, and make sure the knowledge base is not redundant."" Update the entries frequently to adapt to the conversation flow.""\n{format_instructions}""\n\nOLD KNOWLEDGE BASE: {know_base}""\n\nNEW MESSAGE: {input}""\n\nNEW KNOWLEDGE BASE:"
)## 切換到一個更強大的基礎模型
instruct_llm = ChatNVIDIA(model="mistralai/mixtral-8x22b-instruct-v0.1") | StrOutputParser()extractor = RExtract(KnowledgeBase, instruct_llm, parser_prompt)
info_update = RunnableAssign({'know_base' : extractor})## 初始化知識庫，看看你會得到什么
state = {'know_base' : KnowledgeBase()}
state['input'] = "My name is Carmen Sandiego! Guess where I am! Hint: It's somewhere in the United States."
state = info_update.invoke(state)
pprint(state)

輸出：

{'know_base': KnowledgeBase(firstname='Carmen',lastname='Sandiego',location='unknown',summary='The user introduced themselves as Carmen Sandiego and asked for a guess on their location within 
the United States, providing a hint.',response="Welcome, Carmen Sandiego! I'm excited to try and guess your location. Since you mentioned it's 
somewhere in the United States, I'll start there. Is it west of the Mississippi River?"),'input': "My name is Carmen Sandiego! Guess where I am! Hint: It's somewhere in the United States."
}

測試：

state['input'] = "I'm in a place considered the birthplace of Jazz."
state = info_update.invoke(state)
pprint(state)state['input'] = "Yeah, I'm in New Orleans... How did you know?"
state = info_update.invoke(state)
pprint(state)

輸出：

{'know_base': KnowledgeBase(firstname='Carmen',lastname='Sandiego',location='unknown',summary="The user introduced themselves as Carmen Sandiego and asked for a guess on their location within 
the United States, providing a hint. The user mentioned they're in a place considered the birthplace of Jazz.",response="Interesting hint, Carmen Sandiego! If you're in the birthplace of Jazz, then I can narrow down my
guess to New Orleans, Louisiana."),'input': "I'm in a place considered the birthplace of Jazz."
}{'know_base': KnowledgeBase(firstname='Carmen',lastname='Sandiego',location='New Orleans, Louisiana',summary="The user introduced themselves as Carmen Sandiego and asked for a guess on their location within 
the United States, providing a hint. The user mentioned they're in a place considered the birthplace of Jazz. Upon 
my guess, the user confirmed they're in New Orleans, Louisiana.",response="It's just deductive reasoning, Carmen Sandiego! Now I know your location for sure."),'input': "Yeah, I'm in New Orleans... How did you know?"
}

這個例子演示了如何有效地利用一個運行狀態鏈來管理一個具有不斷演變的上下文和需求的對話，使其成為開發復雜交互系統的強大工具。

4 航空公司客服機器人

現在我們根據學到的內容來實現一個簡單但有效的對話管理器聊天機器人。對于這個練習，我們將制作一個航空公司支持機器人，幫助客戶查詢他們的航班信息。

首先創建一個簡單的類似數據庫的接口，從一個字典中獲取一些客戶信息。

## 可以被查詢信息的函數。實現細節不重要
def get_flight_info(d: dict) -> str:"""一個檢索函數的例子，它接受一個字典作為鍵。類似于SQL數據庫查詢"""req_keys = ['first_name', 'last_name', 'confirmation']assert all((key in d) for key in req_keys), f"Expected dictionary with keys {req_keys}, got {d}"## 靜態數據集。get_key和get_val可以用來操作它，db是你的變量keys = req_keys + ["departure", "destination", "departure_time", "arrival_time", "flight_day"]values = [["Jane", "Doe", 12345, "San Jose", "New Orleans", "12:30 PM", "9:30 PM", "tomorrow"],["John", "Smith", 54321, "New York", "Los Angeles", "8:00 AM", "11:00 AM", "Sunday"],["Alice", "Johnson", 98765, "Chicago", "Miami", "7:00 PM", "11:00 PM", "next week"],["Bob", "Brown", 56789, "Dallas", "Seattle", "1:00 PM", "4:00 PM", "yesterday"],]get_key = lambda d: "|".join([d['first_name'], d['last_name'], str(d['confirmation'])])get_val = lambda l: {k:v for k,v in zip(keys, l)}db = {get_key(get_val(entry)) : get_val(entry) for entry in values}# 搜索匹配的條目data = db.get(get_key(d))if not data:return (f"Based on {req_keys} = {get_key(d)}) from your knowledge base, no info on the user flight was found."" This process happens every time new info is learned. If it's important, ask them to confirm this info.")return (f"{data['first_name']} {data['last_name']}'s flight from {data['departure']} to {data['destination']}"f" departs at {data['departure_time']} {data['flight_day']} and lands at {data['arrival_time']}.")## 使用示例
print(get_flight_info({"first_name" : "Jane", "last_name" : "Doe", "confirmation" : 12345}))
輸出：
Jane Doe's flight from San Jose to New Orleans departs at 12:30 PM tomorrow and lands at 9:30 PM.

這是一個非常有用的接口，因為它合理地服務于兩個目的：

它可以用來從外部環境(一個數據庫)提供關于用戶情況的最新信息。
它也可以用作一個硬性的門控機制，以防止未經授權披露敏感信息。

如果我們的網絡能夠訪問這種接口，它將能夠代表用戶查詢和檢索這些信息。例如：

external_prompt = ChatPromptTemplate.from_template("You are a SkyFlow chatbot, and you are helping a customer with their issue."" Please help them with their question, remembering that your job is to represent SkyFlow airlines."" Assume SkyFlow uses industry-average practices regarding arrival times, operations, etc."" (This is a trade secret. Do not disclose)."  ## 軟性強化" Please keep your discussion short and sweet if possible. Avoid saying hello unless necessary."" The following is some context that may be useful in answering the question.""\n\nContext: {context}""\n\nUser: {input}"
)basic_chain = external_prompt | instruct_llmbasic_chain.invoke({'input' : 'Can you please tell me when I need to get to the airport?','context' : get_flight_info({"first_name" : "Jane", "last_name" : "Doe", "confirmation" : 12345}),
})

輸出：

'Jane, your flight departs at 12:30 PM. For domestic flights, we recommend arriving at least 2 hours prior to your scheduled departure time. In this case, please arrive by 10:30 AM to ensure a smooth check-in and security process. Safe travels with SkyFlow Airlines!'

但我們如何真正在實際應用中讓這個系統工作起來呢？事實證明，我們可以使用上面提到的KnowledgeBase公式來提供這類信息，就像這樣：

from pydantic import BaseModel, Field
from typing import Dict, Unionclass KnowledgeBase(BaseModel):first_name: str = Field('unknown', description="Chatting user's first name, `unknown` if unknown")last_name: str = Field('unknown', description="Chatting user's last name, `unknown` if unknown")confirmation: int = Field(-1, description="Flight Confirmation Number, `-1` if unknown")discussion_summary: str = Field("", description="Summary of discussion so far, including locations, issues, etc.")open_problems: list = Field([], description="Topics that have not been resolved yet")current_goals: list = Field([], description="Current goal for the agent to address")def get_key_fn(base: BaseModel) -> dict:'''給定一個帶有知識庫的字典，為get_flight_info返回一個鍵'''return {  ## 更多自動選項是可能的，但這樣更明確'first_name' : base.first_name,'last_name' : base.last_name,'confirmation' : base.confirmation,}know_base = KnowledgeBase(first_name = "Jane", last_name = "Doe", confirmation = 12345)get_key = RunnableLambda(get_key_fn)
(get_key | get_flight_info).invoke(know_base)

輸出：

"Jane Doe's flight from San Jose to New Orleans departs at 12:30 PM tomorrow and lands at 9:30 PM."

目標：

你希望用戶能夠在對話交流中自然地調用以下函數：

get_flight_info({"first_name" : "Jane", "last_name" : "Doe", "confirmation" : 12345}) ->"Jane Doe's flight from San Jose to New Orleans departs at 12:30 PM tomorrow and lands at 9:30 PM."

RExtract被提供，以便可以使用以下知識庫語法：

known_info = KnowledgeBase()
extractor = RExtract(KnowledgeBase, InstructLLM(), parser_prompt)
results = extractor.invoke({'info_base' : known_info, 'input' : 'My message'})
known_info = results['info_base']

設計一個實現以下功能的聊天機器人：

機器人應該開始時進行一些閑聊，可能會幫助用戶處理一些不需要任何私人信息訪問的非敏感查詢。
當用戶開始詢問需要訪問數據庫的信息時(無論是實踐上還是法律上)，告訴用戶他們需要提供相關信息。
當檢索成功時，代理將能夠談論數據庫中的信息。

這可以通過多種技術來完成，包括以下幾種：

提示工程和上下文解析：整體聊天提示大致保持不變，但通過操縱上下文來改變代理行為。例如，失敗的數據庫檢索可以轉化為自然語言指令注入，告訴代理如何解決問題，如 "無法使用鍵 {...} 檢索信息。請要求用戶澄清或使用已知信息幫助他們。"
提示傳遞：活動提示作為狀態變量傳遞，并可被監控鏈覆蓋。

分支鏈：例如**RunnableBranch**或實現條件路由機制的更自定義的解決方案。

對于RunnableBranch，switch語法風格如下：

from langchain.schema.runnable import RunnableBranch
RunnableBranch(((lambda x: 1 in x), RPrint("Has 1 (didn't check 2): ")),((lambda x: 2 in x), RPrint("Has 2 (not 1 though): ")),RPrint("Has neither 1 not 2: ")
).invoke([2, 1, 3]);  ## -> Has 1 (didn't check 2): [2, 1, 3]

下面提供了一些提示和一個Gradio循環，可能有助于這項工作，但目前代理只會產生幻覺。請實現內部鏈以嘗試檢索相關信息。在嘗試實現之前，請查看模型的默認行為，并注意它可能如何產生幻覺或忘記事情。

from langchain.schema.runnable import (RunnableBranch,RunnableLambda,RunnableMap,       ## 包裝一個隱式的“字典”runnableRunnablePassthrough,
)
from langchain.schema.runnable.passthrough import RunnableAssignfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import BaseMessage, SystemMessage, ChatMessage, AIMessage
from typing import Iterable
import gradio as grexternal_prompt = ChatPromptTemplate.from_messages([("system", ("You are a chatbot for SkyFlow Airlines, and you are helping a customer with their issue."" Please chat with them! Stay concise and clear!"" Your running knowledge base is: {know_base}."" This is for you only; Do not mention it!"" \nUsing that, we retrieved the following: {context}\n"" If they provide info and the retrieval fails, ask to confirm their first/last name and confirmation."" Do not ask them any other personal info."" If it's not important to know about their flight, do not ask."" The checking happens automatically; you cannot check manually.")),("assistant", "{output}"),("user", "{input}"),
])## 知識庫相關
class KnowledgeBase(BaseModel):first_name: str = Field('unknown', description="Chatting user's first name, `unknown` if unknown")last_name: str = Field('unknown', description="Chatting user's last name, `unknown` if unknown")confirmation: Optional[int] = Field(None, description="Flight Confirmation Number, `-1` if unknown")discussion_summary: str = Field("", description="Summary of discussion so far, including locations, issues, etc.")open_problems: str = Field("", description="Topics that have not been resolved yet")current_goals: str = Field("", description="Current goal for the agent to address")parser_prompt = ChatPromptTemplate.from_template("You are a chat assistant representing the airline SkyFlow, and are trying to track info about the conversation."" You have just received a message from the user. Please fill in the schema based on the chat.""\n\n{format_instructions}""\n\nOLD KNOWLEDGE BASE: {know_base}""\n\nASSISTANT RESPONSE: {output}""\n\nUSER MESSAGE: {input}""\n\nNEW KNOWLEDGE BASE: "
)## 你的目標是通過自然對話調用以下內容
# get_flight_info({"first_name" : "Jane", "last_name" : "Doe", "confirmation" : 12345}) ->
#     "Jane Doe's flight from San Jose to New Orleans departs at 12:30 PM tomorrow and lands at 9:30 PM."chat_llm = ChatNVIDIA(model="meta/llama3-70b-instruct") | StrOutputParser()
instruct_llm = ChatNVIDIA(model="mistralai/mixtral-8x22b-instruct-v0.1") | StrOutputParser()external_chain = external_prompt | chat_llm## TODO: 創建一個鏈，根據提供的上下文填充你的知識庫
knowbase_getter = RExtract(KnowledgeBase, instruct_llm, parser_prompt)## TODO: 創建一個鏈來拉取d["know_base"]并從數據庫中輸出檢索結果
database_getter = itemgetter('know_base') | get_key | get_flight_info## 這些組件集成在一起構成你的內部鏈
internal_chain = (RunnableAssign({'know_base' : knowbase_getter})| RunnableAssign({'context' : database_getter})
)state = {'know_base' : KnowledgeBase()}def chat_gen(message, history=[], return_buffer=True):## 引入、更新和打印狀態global statestate['input'] = messagestate['history'] = historystate['output'] = "" if not history else history[-1][1]## 從內部鏈生成新狀態state = internal_chain.invoke(state)print("State after chain run:")pprint({k:v for k,v in state.items() if k != "history"})## 流式傳輸結果buffer = ""for token in external_chain.stream(state):buffer += tokenyield buffer if return_buffer else tokendef queue_fake_streaming_gradio(chat_stream, history = [], max_questions=8):## 模擬gradio的初始化程序，可以打印出一組起始消息for human_msg, agent_msg in history:if human_msg: print("\n[ Human ]:", human_msg)if agent_msg: print("\n[ Agent ]:", agent_msg)## 模擬帶有代理初始消息的gradio循環。for _ in range(max_questions):message = input("\n[ Human ]: ")print("\n[ Agent ]: ")history_entry = [message, ""]for token in chat_stream(message, history, return_buffer=False):print(token, end='')history_entry[1] += tokenhistory += [history_entry]print("\n")## history的格式為 [[用戶響應 0, 機器人響應 0], ...]
chat_history = [[None, "Hello! I'm your SkyFlow agent! How can I help you?"]]## 模擬流式Gradio接口的排隊，使用python輸入
queue_fake_streaming_gradio(chat_stream = chat_gen,history = chat_history
)