亚洲欧美视频一区,嘿咻视频免费网站,国产精品久久久久久亚洲伦理

2.1 純代碼架構

下面的基于代碼的代理由 OpenAI 驅動的路由器組成，該路由器使用函數(shù)調(diào)用來選擇要使用的正確技能。該技能完成后，它會返回路由器以調(diào)用另一項技能或響應用戶。

代理會保存一個持續(xù)的消息和響應列表，這些列表在每次調(diào)用時都會完全傳遞到路由器中，以在整個周期中保留上下文。

def router(messages):



if not any(



isinstance(message, dict) and message.get("role") == "system" for message in messages



):



system_prompt = {"role": "system", "content": SYSTEM_PROMPT}



messages.append(system_prompt)



response = client.chat.completions.create(



model="gpt-4o",



messages=messages,



tools=skill_map.get_combined_function_description_for_openai(),



)



messages.append(response.choices[0].message)



tool_calls = response.choices[0].message.tool_calls



if tool_calls:



handle_tool_calls(tool_calls, messages)



return router(messages)



else:



return response.choices[0].message.content

技能本身在自己的類（例如 GenerateSQLQuery）中定義，這些類共同保存在 SkillMap 中。路由器本身只與 SkillMap 交互，它使用 SkillMap 加載技能名稱、描述和可調(diào)用函數(shù)。這種方法意味著向代理添加新技能就像將該技能寫為其自己的類，然后將其添加到 SkillMap 中的技能列表中一樣簡單。這里的想法是讓添加新技能變得容易，而不會干擾路由器代碼。

class SkillMap:



def __init__(self):



skills = [AnalyzeData(), GenerateSQLQuery()]



self.skill_map = {}



for skill in skills:



self.skill_map[skill.get_function_name()] = (



skill.get_function_dict(),



skill.get_function_callable(),



)



def get_function_callable_by_name(self, skill_name) -> Callable:



return self.skill_map[skill_name][1]



def get_combined_function_description_for_openai(self):



combined_dict = []



for _, (function_dict, _) in self.skill_map.items():



combined_dict.append(function_dict)



return combined_dict



def get_function_list(self):



return list(self.skill_map.keys())



def get_list_of_function_callables(self):



return [skill[1] for skill in self.skill_map.values()]



def get_function_description_by_name(self, skill_name):



return str(self.skill_map[skill_name][0]["function"])

總體而言，這種方法實施起來相當簡單，但也面臨一些挑戰(zhàn)。

2.2 純代碼代理的挑戰(zhàn)

第一個困難在于構造路由器系統(tǒng)提示。通常，上面示例中的路由器堅持自己生成 SQL，而不是將其委托給正確的技能。如果您曾經(jīng)嘗試讓 LLM 不做某事，您就會知道這種體驗有多么令人沮喪；找到一個有效的提示需要經(jīng)過多輪調(diào)試。考慮每個步驟的不同輸出格式也很棘手。由于我選擇不使用結構化輸出，因此我必須準備好應對路由器和技能中每個 LLM 調(diào)用的多種不同格式。

2.3 純代碼代理的好處

基于代碼的方法提供了良好的基線和起點，提供了一種很好的方法來學習代理的工作原理，而無需依賴來自主流框架的固定代理教程。雖然說服 LLM 表現(xiàn)可能具有挑戰(zhàn)性，但代碼結構本身足夠簡單易用，可能對某些用例有意義（更多信息請參見下面的分析部分）。

3、LangGraph

LangGraph 是最古老的代理框架之一，于 2024 年 1 月首次發(fā)布。該框架旨在通過采用 Pregel 圖結構來解決現(xiàn)有管道和鏈的非循環(huán)性質。LangGraph 通過添加節(jié)點、邊和條件邊的概念來遍歷圖，使在代理中定義循環(huán)變得更加容易。LangGraph 建立在?LangChain?之上，并使用該框架中的對象和類型。

3.1 LangGraph 架構

LangGraph 代理在紙面上看起來與基于代碼的代理類似，但其背后的代碼卻截然不同。LangGraph 在技術上仍然使用“路由器”，因為它使用函數(shù)調(diào)用 OpenAI 并使用響應繼續(xù)執(zhí)行新步驟。然而，程序在技能之間移動的方式控制方式完全不同。

tools = [generate_and_run_sql_query, data_analyzer]



model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)



def create_agent_graph():



workflow = StateGraph(MessagesState)



tool_node = ToolNode(tools)



workflow.add_node("agent", call_model)



workflow.add_node("tools", tool_node)



workflow.add_edge(START, "agent")



workflow.add_conditional_edges(



"agent",



should_continue,



)



workflow.add_edge("tools", "agent")



checkpointer = MemorySaver()



app = workflow.compile(checkpointer=checkpointer)



return app

這里定義的圖表有一個用于初始 OpenAI 調(diào)用的節(jié)點，稱為上面的“代理”，還有一個用于工具處理步驟的節(jié)點，稱為“工具”。LangGraph 有一個名為 ToolNode 的內(nèi)置對象，它獲取可調(diào)用工具的列表并根據(jù) ChatMessage 響應觸發(fā)它們，然后再次返回到“代理”節(jié)點。

def should_continue(state: MessagesState):



messages = state["messages"]



last_message = messages[-1]



if last_message.tool_calls:



return "tools"



return END



def call_model(state: MessagesState):



messages = state["messages"]



response = model.invoke(messages)



return {"messages": [response]}

每次調(diào)用“代理”節(jié)點（換句話說：基于代碼的代理中的路由器）后，should_continue 邊緣都會決定是否將響應返回給用戶或傳遞給 ToolNode 來處理工具調(diào)用。

在每個節(jié)點中，“狀態(tài)”存儲來自 OpenAI 的消息和響應列表，類似于基于代碼的代理的方法。

3.2 LangGraph 的挑戰(zhàn)

示例中 LangGraph 的大部分困難源于需要使用 Langchain 對象才能順利運行。

挑戰(zhàn) 1：函數(shù)調(diào)用驗證

為了使用 ToolNode 對象，我不得不重構大部分現(xiàn)有的 Skill 代碼。ToolNode 采用可調(diào)用函數(shù)列表，這最初讓我認為我可以使用現(xiàn)有的函數(shù)，但由于我的函數(shù)參數(shù)，事情變得一團糟。

技能被定義為具有可調(diào)用成員函數(shù)的類，這意味著它們的第一個參數(shù)是“self”。GPT-4o 非常聰明，不會在生成的函數(shù)調(diào)用中包含“self”參數(shù)，但 LangGraph 將其解讀為由于缺少參數(shù)而導致的驗證錯誤。

這花了幾個小時才弄清楚，因為錯誤消息反而將函數(shù)中的第三個參數(shù)（數(shù)據(jù)分析技能中的“args”）標記為缺少的參數(shù)：

pydantic.v1.error_wrappers.ValidationError: 1 validation error for data_analysis_toolSchema
args field required (type=value_error.missing)

值得一提的是，錯誤信息來自 Pydantic，而不是 LangGraph。

我最終下定決心，用 Langchain 的 @tool 裝飾器將我的技能重新定義為基本方法，并讓一切正常運轉。

@tool



def generate_and_run_sql_query(query: str):



"""Generates and runs an SQL query based on the prompt.



Args:



query (str): A string containing the original user prompt.



Returns:



str: The result of the SQL query.



"""

挑戰(zhàn) #2：調(diào)試

如前所述，在框架中調(diào)試很困難。這主要歸結于令人困惑的錯誤消息和抽象概念，這些使查看變量變得更加困難。

抽象概念主要出現(xiàn)在嘗試調(diào)試代理周圍發(fā)送的消息時。LangGraph 將這些消息存儲在狀態(tài) [“messages”] 中。圖中的某些節(jié)點會自動從這些消息中提取數(shù)據(jù)，這可能會使節(jié)點訪問消息時難以理解消息的值。

3.3 LangGraph 的優(yōu)勢

LangGraph 的主要優(yōu)勢之一是易于使用。圖形結構代碼簡潔易懂。特別是如果您有復雜的節(jié)點邏輯，擁有圖形的單一視圖可以更輕松地理解代理是如何連接在一起的。LangGraph 還可以輕松轉換在 LangChain 中構建的現(xiàn)有應用程序。

3.4 總結

如果您使用框架中的所有內(nèi)容，LangGraph 可以干凈利落地運行；如果您不使用它，請準備好面對一些調(diào)試難題。

4、LlamaIndex Workflows

Workflows 是代理框架領域的新成員，于今年夏初首次亮相。與 LangGraph 一樣，它旨在使循環(huán)代理更易于構建。Workflows 還特別注重異步運行。

Workflows 的一些元素似乎直接響應了 LangGraph，特別是它使用事件而不是邊和條件邊。Workflows 使用步驟（類似于 LangGraph 中的節(jié)點）來容納邏輯，并使用發(fā)出和接收事件在步驟之間移動。

上面的結構看起來與 LangGraph 結構類似，除了一個附加項。我在 Workflow 中添加了一個設置步驟來準備代理上下文，下面將詳細介紹。盡管結構相似，但支持它的代碼卻大不相同。

4.1 Workflows 架構

下面的代碼定義了 Workflow 結構。與 LangGraph 類似，這是我準備狀態(tài)并將技能附加到 LLM 對象的地方。

class AgentFlow(Workflow):



def __init__(self, llm, timeout=300):



super().__init__(timeout=timeout)



self.llm = llm



self.memory = ChatMemoryBuffer(token_limit=1000).from_defaults(llm=llm)



self.tools = []



for func in skill_map.get_function_list():



self.tools.append(



FunctionTool(



skill_map.get_function_callable_by_name(func),



metadata=ToolMetadata(



name=func, description=skill_map.get_function_description_by_name(func)



),



)



)



@step



async def prepare_agent(self, ev: StartEvent) -> RouterInputEvent:



user_input = ev.input



user_msg = ChatMessage(role="user", content=user_input)



self.memory.put(user_msg)



chat_history = self.memory.get()



return RouterInputEvent(input=chat_history)

這也是我定義額外步驟“prepare_agent”的地方。此步驟根據(jù)用戶輸入創(chuàng)建一個 ChatMessage 并將其添加到工作流內(nèi)存中。將其拆分為單獨的步驟意味著我們在代理循環(huán)執(zhí)行步驟時會返回到它，從而避免反復將用戶消息添加到內(nèi)存中。

在 LangGraph 案例中，我使用位于圖表之外的 run_agent 方法完成了同樣的事情。這種變化主要是風格上的，但我認為像我們在這里所做的那樣，將此邏輯與工作流和圖表放在一起會更簡潔。

設置好工作流后，我定義了路由代碼：

@step



async def router(self, ev: RouterInputEvent) -> ToolCallEvent | StopEvent:



messages = ev.input



if not any(



isinstance(message, dict) and message.get("role") == "system" for message in messages



):



system_prompt = ChatMessage(role="system", content=SYSTEM_PROMPT)



messages.insert(0, system_prompt)



with using_prompt_template(template=SYSTEM_PROMPT, version="v0.1"):



response = await self.llm.achat_with_tools(



model="gpt-4o",



messages=messages,



tools=self.tools,



)



self.memory.put(response.message)



tool_calls = self.llm.get_tool_calls_from_response(response, error_on_no_tool_call=False)



if tool_calls:



return ToolCallEvent(tool_calls=tool_calls)



else:



return StopEvent(result=response.message.content)

以及工具調(diào)用處理代碼：

@step



async def tool_call_handler(self, ev: ToolCallEvent) -> RouterInputEvent:



tool_calls = ev.tool_calls



for tool_call in tool_calls:



function_name = tool_call.tool_name



arguments = tool_call.tool_kwargs



if "input" in arguments:



arguments["prompt"] = arguments.pop("input")



try:



function_callable = skill_map.get_function_callable_by_name(function_name)



except KeyError:



function_result = "Error: Unknown function call"



function_result = function_callable(arguments)



message = ChatMessage(



role="tool",



content=function_result,



additional_kwargs={"tool_call_id": tool_call.tool_id},



)



self.memory.put(message)



return RouterInputEvent(input=self.memory.get())

這兩個看起來更像基于代碼的代理而不是 LangGraph 代理。這主要是因為 Workflows 將條件路由邏輯保留在步驟中而不是條件邊緣中 — 第 18-24 行是 LangGraph 中的條件邊緣，而現(xiàn)在它們只是路由步驟的一部分 — 并且 LangGraph 有一個 ToolNode 對象，它幾乎自動執(zhí)行 tool_call_handler 方法中的所有操作。

經(jīng)過路由步驟后，我很高興看到一件事，那就是我可以將我的 SkillMap 和基于代碼的代理中的現(xiàn)有技能與 Workflows 結合使用。這些不需要任何更改即可與 Workflows 配合使用，這讓我的生活變得輕松多了。

4.2 Workflows 的挑戰(zhàn)

挑戰(zhàn) #1：同步與異步

雖然異步執(zhí)行對于實時代理來說是更好的選擇，但調(diào)試同步代理要容易得多。Workflows 旨在異步工作，嘗試強制同步執(zhí)行非常困難。

我最初以為我只需刪除“異步”方法標識，并從“achat_with_tools”切換到“chat_with_tools”即可。但是，由于 Workflow 類中的底層方法也被標記為異步，因此必須重新定義這些方法才能同步運行。我最終堅持使用異步方法，但這并沒有使調(diào)試變得更加困難。

挑戰(zhàn) #2：Pydantic 驗證錯誤

與 LangGraph 的困境如出一轍，類似的問題也出現(xiàn)在技能上令人困惑的 Pydantic 驗證錯誤。幸運的是，這一次這些問題更容易解決，因為 Workflows 能夠很好地處理成員函數(shù)。最終，我不得不更加規(guī)范地為我的技能創(chuàng)建 LlamaIndex FunctionTool 對象：

for func in skill_map.get_function_list():
self.tools.append(FunctionTool(
skill_map.get_function_callable_by_name(func),
metadata=ToolMetadata(name=func, description=skill_map.get_function_description_by_name(func))))

4.3 工作流的優(yōu)勢

與 LangGraph 代理相比，我構建工作流代理要容易得多，主要是因為工作流仍然需要我自己編寫路由邏輯和工具處理代碼，而不是提供內(nèi)置函數(shù)。這也意味著我的工作流代理看起來與我的基于代碼的代理極為相似。

最大的區(qū)別在于事件的使用。我使用兩個自定義事件在代理中的步驟之間移動：

class ToolCallEvent(Event):
tool_calls: list[ToolSelection]
class RouterInputEvent(Event):
input: list[ChatMessage]

基于事件的發(fā)射器-接收器架構取代了直接調(diào)用我的代理中的某些方法，例如工具調(diào)用處理程序。

如果您擁有更復雜的系統(tǒng)，其中包含多個異步觸發(fā)的步驟，并且可能會發(fā)出多個事件，那么這種架構對于干凈地管理這些步驟非常有用。

工作流的其他好處包括它非常輕量級，不會強迫您使用太多結構（除了使用某些 LlamaIndex 對象），并且其基于事件的架構為直接函數(shù)調(diào)用提供了一種有用的替代方案——尤其是對于復雜的異步應用程序。

4.4 比較框架

縱觀這三種方法，每種方法都有其優(yōu)點。

無框架方法是最容易實現(xiàn)的。因為任何抽象都是由開發(fā)人員定義的（即上例中的 SkillMap 對象），所以保持各種類型和對象的連貫性很容易。然而，代碼的可讀性和可訪問性完全取決于個人開發(fā)人員，很容易看出，如果沒有一些強制結構，日益復雜的代理會變得多么混亂。

LangGraph 提供了相當多的結構，這使得代理的定義非常明確。如果一個更廣泛的團隊正在合作開發(fā)代理，這種結構將提供一種實施架構的有用方法。對于那些不太熟悉該結構的人來說，LangGraph 也可能是一個很好的代理起點。然而，這有一個權衡——由于 LangGraph 為您做了很多事情，如果您沒有完全接受該框架，它可能會導致麻煩；代碼可能非常干凈，但您可能需要付出更多調(diào)試的代價。

工作流介于兩者之間。基于事件的架構可能對某些項目非常有用，而且在使用 LlamaIndex 類型方面要求更少的事實為那些沒有在其應用程序中完全使用框架的人提供了更大的靈活性。