AI Agents-5 | AI工作流與代理的深度剖析：從原理到實戰應用 原創
這個系列文章旨在為AI代理（AI Agent）提供全面的概述，深入研究其特征，組成部分和類型，同時探索其進化，挑戰和潛在的未來方向。
在人工智能飛速發展的今天，我們常常聽到“AI工作流”和“AI代理”這兩個概念。但你知道它們之間有什么區別嗎？又該如何選擇適合自己的技術路徑呢？今天，我們就來深入探討一下。
一、工作流與代理：到底有什么區別？
（一）定義與區別
在AI的世界里，“代理”（Agent）和“工作流”（Workflow）是兩種截然不同的存在。有些人把代理定義為完全自主的系統，它們可以獨立運行很長時間，利用各種工具完成復雜的任務。而另一些人則用它來描述那些按照預定義流程執行的系統。
Anthropic公司把這兩種情況都歸類為“代理系統”，但他們明確區分了工作流和代理：
- 工作流：是一種通過預定義代碼路徑來協調LLM（大型語言模型）和工具的系統。
- 代理：則是LLM動態地指導自己的流程和工具使用，完全掌控任務的執行方式。
用一個簡單的比喻來說，工作流就像是按照食譜一步步做飯，而代理則像是一個廚師，根據食材和口味現場決定怎么做菜。
（二）什么時候用代理，什么時候不用？
開發基于LLM的應用時，最好從最簡單的解決方案開始，只有在必要時才引入復雜性。有時候，完全避開代理系統可能是個更好的選擇。畢竟，這些系統雖然能提高任務性能，但往往伴隨著更高的延遲和成本。所以，我們需要權衡利弊。
如果任務結構清晰、規則明確，工作流可以提供穩定性和一致性。而代理則更適合那些需要靈活性和大規模模型驅動決策的場景。不過，對于很多應用來說，優化單個LLM調用，加上檢索和上下文示例，往往就足夠了。
（三）框架的使用：利與弊
現在有很多框架可以讓代理系統的實現變得更簡單，比如LangChain的LangGraph、亞馬遜Bedrock的AI代理框架、Rivet（一個拖拽式GUI的LLM工作流構建器）和Vellum（用于構建和測試復雜工作流的GUI工具）。這些框架簡化了調用LLM、定義和解析工具、串聯調用等標準低級任務，讓你更容易上手。
但這些框架也有缺點。它們可能會增加額外的抽象層，讓你看不清底層的提示和響應，從而讓調試變得困難。而且，它們可能會讓你不自覺地增加復雜性，而實際上簡單的設置就足夠了。
所以，我們建議開發者先直接使用LLM的API，因為很多模式只需要幾行代碼就能實現。如果你決定使用框架，一定要徹底理解底層代碼，因為對框架內部工作原理的錯誤假設是錯誤的常見來源。
二、構建模塊：從簡單到復雜
（一）基礎設置
在構建代理系統時，你可以使用任何支持結構化輸出和工具調用的聊天模型。以下是一個簡單的設置過程，展示了如何安裝包、設置API密鑰，并測試Anthropic的結構化輸出和工具調用。
import os
import getpass
from langchain_anthropic import ChatAnthropic
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("ANTHROPIC_API_KEY")
llm = ChatAnthropic(model="claude-3-5-sonnet-latest")（二）增強型LLM：核心構建模塊
代理系統的基礎是一個增強型LLM，它通過檢索、工具和記憶等增強功能來實現更強大的能力。我們的模型可以主動利用這些能力——生成自己的搜索查詢、選擇合適的工具、決定保留哪些信息。
我們建議重點關注兩個方面：一是將這些能力定制化到你的具體用例中；二是確保它們為LLM提供一個易于使用且文檔齊全的接口。雖然實現這些增強功能有很多方法，但一種方法是通過我們最近發布的模型上下文協議（Model Context Protocol），它允許開發者通過簡單的客戶端實現與不斷增長的第三方工具生態系統集成。
從現在開始，我們假設每個LLM調用都能訪問這些增強功能。
（三）工作流的實現
1. 鏈式提示（Prompt Chaining）
鏈式提示是一種將任務分解為一系列步驟的工作流，每個LLM調用都處理前一個的輸出。你可以在這個過程中加入程序化檢查（見下圖中的“門”），以確保流程仍在正軌上。
這種工作流適用于那些可以輕松分解為固定子任務的場景。主要目標是通過增加延遲來換取更高的準確性，讓每個LLM調用的任務變得更簡單。
例如：
- 先生成營銷文案，然后將其翻譯成另一種語言。
- 先寫文檔大綱，檢查大綱是否符合某些標準，再根據大綱撰寫文檔。
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display
# 圖狀態
class State(TypedDict):
topic: str
joke: str
improved_joke: str
final_joke: str
# 節點
def generate_joke(state: State):
"""第一次LLM調用，生成初始笑話"""
msg = llm.invoke(f"Write a short joke about {state['topic']}")
return {"joke": msg.content}
def check_punchline(state: State):
"""檢查笑話是否有笑點"""
if"?"in state["joke"] or"!"in state["joke"]:
return"Fail"
return"Pass"
def improve_joke(state: State):
"""第二次LLM調用，改進笑話"""
msg = llm.invoke(f"Make this joke funnier by adding wordplay: {state['joke']}")
return {"improved_joke": msg.content}
def polish_joke(state: State):
"""第三次LLM調用，完善笑話"""
msg = llm.invoke(f"Add a surprising twist to this joke: {state['improved_joke']}")
return {"final_joke": msg.content}
# 構建工作流
workflow = StateGraph(State)
# 添加節點
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)
# 添加邊連接節點
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
"generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)
# 編譯
chain = workflow.compile()
# 顯示工作流
display(Image(chain.get_graph().draw_mermaid_png()))
# 調用
state = chain.invoke({"topic": "cats"})
print("初始笑話：")
print(state["joke"])
print("\n--- --- ---\n")
if"improved_joke"in state:
print("改進后的笑話：")
print(state["improved_joke"])
print("\n--- --- ---\n")
print("最終笑話：")
print(state["final_joke"])
else:
print("笑話未通過質量檢查——未檢測到笑點！")2. 并行化（Parallelization）
并行化是一種讓LLM同時處理任務的工作流，其輸出可以通過程序化的方式聚合。這種工作流有兩種主要形式：分段（Sectioning）和投票（Voting）。
- 分段：將任務分解為獨立的子任務并并行運行。
- 投票：多次運行相同的任務以獲得多樣化的輸出。
這種工作流適用于以下場景：當子任務可以并行化以提高速度時，或者需要多種視角或嘗試以獲得更可靠的結果時。對于復雜的任務，如果每個考慮因素都由單獨的LLM調用處理，LLM通常會表現得更好，因為這樣可以讓每個調用專注于特定的方面。
例如：
- 分段：實現防護欄，一個模型實例處理用戶查詢，另一個篩選不當內容或請求。這比讓同一個LLM調用處理防護欄和核心響應表現得更好。
- 投票：審查代碼漏洞，多個不同的提示檢查并標記問題；評估內容是否不當，多個提示從不同角度評估，或需要不同的投票閾值以平衡誤報和漏報。
# 圖狀態
class State(TypedDict):
topic: str
joke: str
story: str
poem: str
combined_output: str
# 節點
def call_llm_1(state: State):
"""第一次LLM調用，生成笑話"""
msg = llm.invoke(f"Write a joke about {state['topic']}")
return {"joke": msg.content}
def call_llm_2(state: State):
"""第二次LLM調用，生成故事"""
msg = llm.invoke(f"Write a story about {state['topic']}")
return {"story": msg.content}
def call_llm_3(state: State):
"""第三次LLM調用，生成詩歌"""
msg = llm.invoke(f"Write a poem about {state['topic']}")
return {"poem": msg.content}
def aggregator(state: State):
"""將笑話、故事和詩歌合并為一個輸出"""
combined = f"Here's a story, joke, and poem about {state['topic']}!\n\n"
combined += f"STORY:\n{state['story']}\n\n"
combined += f"JOKE:\n{state['joke']}\n\n"
combined += f"POEM:\n{state['poem']}"
return {"combined_output": combined}
# 構建工作流
parallel_builder = StateGraph(State)
# 添加節點
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)
# 添加邊連接節點
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
parallel_builder.add_edge("aggregator", END)
# 編譯工作流
parallel_workflow = parallel_builder.compile()
# 顯示工作流
display(Image(parallel_workflow.get_graph().draw_mermaid_png()))
# 調用
state = parallel_workflow.invoke({"topic": "cats"})
print(state["combined_output"])3. 路由（Routing）
路由工作流可以根據輸入的分類將其導向后續任務。這種工作流允許分離關注點，并構建更專業的提示。如果沒有這種工作流，優化一種輸入的性能可能會損害其他輸入的性能。
這種工作流適用于以下場景：當任務復雜且有明確的類別，這些類別可以分別處理，并且分類可以準確完成（無論是通過LLM還是更傳統的分類模型/算法）。
例如：
- 將不同類型的客戶服務查詢（一般問題、退款請求、技術支持）導向不同的下游流程、提示和工具。
- 將簡單/常見問題導向較小的模型（如Claude 3.5 Haiku），將復雜/罕見問題導向更強大的模型（如Claude 3.5 Sonnet），以優化成本和速度。
from typing_extensions import Literal
from langchain_core.messages import HumanMessage, SystemMessage
# 結構化輸出的模式，用于路由邏輯
class Route(BaseModel):
step: Literal["poem", "story", "joke"] = Field(
None, descriptinotallow="路由過程中的下一步"
)
# 為LLM添加結構化輸出模式
router = llm.with_structured_output(Route)
# 狀態
class State(TypedDict):
input: str
decision: str
output: str
# 節點
def llm_call_1(state: State):
"""寫一個故事"""
result = llm.invoke(state["input"])
return {"output": result.content}
def llm_call_2(state: State):
"""寫一個笑話"""
result = llm.invoke(state["input"])
return {"output": result.content}
def llm_call_3(state: State):
"""寫一首詩"""
result = llm.invoke(state["input"])
return {"output": result.content}
def llm_call_router(state: State):
"""根據輸入將任務路由到相應的節點"""
decision = router.invoke(
[
SystemMessage(
cnotallow="根據用戶請求將輸入路由到故事、笑話或詩歌。"
),
HumanMessage(cnotallow=state["input"]),
]
)
return {"decision": decision.step}
# 條件邊函數，根據路由決策將任務導向相應節點
def route_decision(state: State):
if state["decision"] == "story":
return"llm_call_1"
elif state["decision"] == "joke":
return"llm_call_2"
elif state["decision"] == "poem":
return"llm_call_3"
# 構建工作流
router_builder = StateGraph(State)
# 添加節點
router_builder.add_node("llm_call_1", llm_call_1)
router_builder.add_node("llm_call_2", llm_call_2)
router_builder.add_node("llm_call_3", llm_call_3)
router_builder.add_node("llm_call_router", llm_call_router)
# 添加邊連接節點
router_builder.add_edge(START, "llm_call_router")
router_builder.add_conditional_edges(
"llm_call_router",
route_decision,
{
"llm_call_1": "llm_call_1",
"llm_call_2": "llm_call_2",
"llm_call_3": "llm_call_3",
},
)
router_builder.add_edge("llm_call_1", END)
router_builder.add_edge("llm_call_2", END)
router_builder.add_edge("llm_call_3", END)
# 編譯工作流
router_workflow = router_builder.compile()
# 顯示工作流
display(Image(router_workflow.get_graph().draw_mermaid_png()))
# 調用
state = router_workflow.invoke({"input": "Write me a joke about cats"})
print(state["output"])4. 協調者-工作者（Orchestrator-Worker）
在協調者-工作者工作流中，一個中心LLM動態地分解任務，將其分配給工作者LLM，并整合它們的結果。
這種工作流適用于以下場景：當任務復雜且無法預測需要哪些子任務時（例如在編程中，需要更改的文件數量以及每個文件的更改性質通常取決于具體任務）。雖然它在拓撲結構上與并行化類似，但關鍵區別在于其靈活性——子任務不是預先定義的，而是由協調者根據具體輸入動態確定的。
舉個例子，這種工作流非常適合以下場景：
- 編程產品需要對多個文件進行復雜更改，每次更改都可能涉及不同的任務。
- 搜索任務需要從多個來源收集和分析信息，以獲取可能相關的數據。
from typing import Annotated, List
import operator
# 用于規劃的結構化輸出模式
class Section(BaseModel):
name: str = Field(descriptinotallow="報告部分的名稱")
description: str = Field(descriptinotallow="本節涵蓋的主要主題和概念的簡要概述")
class Sections(BaseModel):
sections: List[Section] = Field(descriptinotallow="報告的各個部分")
# 為LLM添加結構化輸出模式
planner = llm.with_structured_output(Sections)
# 創建工作者節點
因為協調者-工作者工作流非常常見，LangGraph提供了Send API來支持這種模式。它允許你動態創建工作者節點，并為每個節點分配特定的輸入。每個工作者都有自己的狀態，并且所有工作者的輸出都會寫入一個共享的狀態鍵，協調者圖可以訪問這個鍵。這使得協調者能夠訪問所有工作者的輸出，并將它們整合成最終輸出。如下所示，我們遍歷一個部分列表，并將每個部分發送到一個工作者節點。
```python
from langgraph.constants import Send
# 圖狀態
class State(TypedDict):
topic: str # 報告主題
sections: list[Section] # 報告部分列表
completed_sections: Annotated[list, operator.add] # 所有工作者并行寫入此鍵
final_report: str # 最終報告
# 工作者狀態
class WorkerState(TypedDict):
section: Section
completed_sections: Annotated[list, operator.add]
# 節點
def orchestrator(state: State):
"""協調者生成報告計劃"""
report_sections = planner.invoke(
[
SystemMessage(cnotallow="生成報告計劃"),
HumanMessage(cnotallow=f"報告主題：{state['topic']}")
]
)
return {"sections": report_sections.sections}
def llm_call(state: WorkerState):
"""工作者撰寫報告的一部分"""
section = llm.invoke(
[
SystemMessage(cnotallow="根據提供的名稱和描述撰寫報告部分。每個部分不加前言，使用Markdown格式。"),
HumanMessage(cnotallow=f"部分名稱：{state['section'].name}，描述：{state['section'].description}")
]
)
return {"completed_sections": [section.content]}
def synthesizer(state: State):
"""將各部分整合成完整報告"""
completed_report_sections = "\n\n---\n\n".join(state["completed_sections"])
return {"final_report": completed_report_sections}
# 條件邊函數，為每個計劃部分分配工作者
def assign_workers(state: State):
"""為計劃中的每個部分分配工作者"""
return [Send("llm_call", {"section": s}) for s in state["sections"]]
# 構建工作流
orchestrator_worker_builder = StateGraph(State)
# 添加節點
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)
# 添加邊連接節點
orchestrator_worker_builder.add_edge(START, "orchestrator")
orchestrator_worker_builder.add_conditional_edges(
"orchestrator", assign_workers, ["llm_call"]
)
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.add_edge("synthesizer", END)
# 編譯工作流
orchestrator_worker = orchestrator_worker_builder.compile()
# 顯示工作流
display(Image(orchestrator_worker.get_graph().draw_mermaid_png()))
# 調用
state = orchestrator_worker.invoke({"topic": "創建關于LLM擴展規律的報告"})
# 顯示最終報告
from IPython.display import Markdown
Markdown(state["final_report"])5. 評估器-優化器（Evaluator-Optimizer）
在評估器-優化器工作流中，一個LLM調用生成響應，另一個提供評估和反饋，并在循環中不斷優化。
這種工作流特別適用于以下場景：當我們有明確的評估標準，并且迭代優化能夠帶來顯著價值時。如果LLM的響應可以在人類明確反饋后得到改進，并且LLM能夠提供這種反饋，那么這種工作流就非常適合。這類似于人類作家在撰寫一篇經過精心打磨的文檔時所經歷的迭代寫作過程。
例如：
- 文學翻譯中，翻譯LLM可能無法在一開始就捕捉到所有細微差別，但評估LLM可以提供有用的批評。
- 復雜的搜索任務需要多輪搜索和分析以收集全面信息，評估器決定是否需要進一步搜索。
# 圖狀態
class State(TypedDict):
joke: str
topic: str
feedback: str
funny_or_not: str
# 用于評估的結構化輸出模式
class Feedback(BaseModel):
grade: Literal["funny", "not funny"] = Field(descriptinotallow="判斷笑話是否有趣")
feedback: str = Field(descriptinotallow="如果笑話不好笑，提供改進建議")
# 為LLM添加結構化輸出模式
evaluator = llm.with_structured_output(Feedback)
# 節點
def llm_call_generator(state: State):
"""LLM生成笑話"""
if state.get("feedback"):
msg = llm.invoke(
f"根據反饋生成關于{state['topic']}的笑話：{state['feedback']}"
)
else:
msg = llm.invoke(f"生成關于{state['topic']}的笑話")
return {"joke": msg.content}
def llm_call_evaluator(state: State):
"""LLM評估笑話"""
grade = evaluator.invoke(f"評估笑話：{state['joke']}")
return {"funny_or_not": grade.grade, "feedback": grade.feedback}
# 條件邊函數，根據評估器的反饋決定是否返回生成器或結束
def route_joke(state: State):
"""根據評估器的反饋決定是否返回生成器或結束"""
if state["funny_or_not"] == "funny":
return"Accepted"
elif state["funny_or_not"] == "not funny":
return"Rejected + Feedback"
# 構建工作流
optimizer_builder = StateGraph(State)
# 添加節點
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)
# 添加邊連接節點
optimizer_builder.add_edge(START, "llm_call_generator")
optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator")
optimizer_builder.add_conditional_edges(
"llm_call_evaluator",
route_joke,
{
"Accepted": END,
"Rejected + Feedback": "llm_call_generator"
}
)
# 編譯工作流
optimizer_workflow = optimizer_builder.compile()
# 顯示工作流
display(Image(optimizer_workflow.get_graph().draw_mermaid_png()))
# 調用
state = optimizer_workflow.invoke({"topic": "Cats"})
print(state["joke"])三、代理（Agent）：自主智能體的力量
代理通常被實現為LLM通過工具調用（基于環境反饋）在循環中執行動作的系統。正如Anthropic博客所指出的，代理可以處理復雜的任務，但其實現往往非常簡單。它們通常是LLM根據環境反饋使用工具的循環。因此，清晰且周到地設計工具集及其文檔至關重要。
什么時候使用代理？
代理適用于那些開放性問題，這些問題很難或無法預測所需的步驟數量，并且無法硬編碼固定路徑。LLM可能會運行多個回合，因此你需要對其決策能力有一定的信任。代理的自主性使其非常適合在受信任的環境中擴展任務。
然而，自主性也意味著更高的成本和可能出現的錯誤累積。因此，我們建議在沙盒環境中進行廣泛的測試，并設置適當的防護欄。
例如：
- 編程代理可以解決涉及對多個文件進行編輯的任務。
- “計算機使用”參考實現中，Claude通過計算機完成任務。
from langchain_core.tools import tool
# 定義工具
@tool
def multiply(a: int, b: int) -> int:
"""乘法工具"""
return a * b
@tool
def add(a: int, b: int) -> int:
"""加法工具"""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""除法工具"""
return a / b
# 為LLM綁定工具
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
llm_with_tools = llm.bind_tools(tools)
四、結合與定制這些模式：實現最佳效果
AI代理和代理工作流是互補的，可以集成在一起以實現最佳效果，尤其是在復雜的現實世界應用中。
（一）增強自動化
AI代理可以自主處理特定任務，而代理工作流則將這些任務協調成一個連貫、高效的過程。
（二）可擴展性
在結構化工作流中結合多個AI代理，可以使組織高效擴展運營，減少人工工作量，提高生產力。
（三）彈性與適應性
雖然單個代理可以應對局部變化，但工作流可以動態調整整體流程，以與戰略目標保持一致或適應外部干擾。
（四）實際案例：制造業中的集成AI代理與工作流
在智能制造系統中：
- AI代理可以監控設備性能、預測維護需求并優化生產計劃。
- 代理工作流則負責原材料采購、生產排序、質量保證和物流，確保從原材料到產品交付的無縫過渡。
五、總結：選擇適合你的系統才是成功的關鍵
在LLM領域，成功并不是關于構建最復雜的系統，而是構建最適合你需求的系統。從簡單的提示開始，通過全面評估進行優化，只有在簡單解決方案不足時才添加多步代理系統。
在實現代理時，我們建議遵循以下三個核心原則：
- 保持代理設計的簡潔性：避免不必要的復雜性，專注于核心功能。
- 優先考慮透明性：明確展示代理的規劃步驟，讓用戶清楚了解其決策過程。
- 精心設計代理-計算機接口（ACI）：通過徹底的工具文檔和測試，確保代理與外部系統的無縫交互。
框架可以幫助你快速上手，但不要害怕在進入生產階段時減少抽象層，直接使用基礎組件。遵循這些原則，你可以創建出不僅強大而且可靠、可維護且值得用戶信賴的代理系統。
在這個充滿可能性的AI時代，選擇合適的技術路徑并將其應用于實際場景，才是實現智能化轉型的關鍵。希望這篇文章能幫助你在AI工作流與代理的世界中找到屬于你的方向，解鎖智能技術的無限可能。
本文轉載自公眾號Halo咯咯 作者：基咯咯
