全網(wǎng)最全AI智能體框架終極指南：從CoT到Plan-and-Execute，附LangGraph完整代碼原創(chuàng)

發(fā)布于 2025-8-27 07:56

瀏覽

1收藏

隨著大語(yǔ)言模型（LLM）的快速發(fā)展，如何讓AI系統(tǒng)更智能地解決復(fù)雜問(wèn)題成為了研究熱點(diǎn)。從簡(jiǎn)單的問(wèn)答到復(fù)雜的多步驟任務(wù)執(zhí)行，LLM Agent思考框架正在重塑我們對(duì)AI能力的認(rèn)知。本文將深入探討幾種主流的Agent思考框架，包括Chain of Thought（CoT）、ReAct、以及Plan-and-Execute等，并提供基于LangChain和LangGraph的具體實(shí)現(xiàn)代碼，幫助您理解這些框架的原理、優(yōu)勢(shì)和實(shí)際應(yīng)用場(chǎng)景。

無(wú)論您是想快速上手Agent開發(fā)，還是希望深入理解不同框架的設(shè)計(jì)理念，本文都將為您提供理論知識(shí)和實(shí)戰(zhàn)代碼的完整指南。

一、Agent思考框架概述

環(huán)境準(zhǔn)備

在開始之前，請(qǐng)確保安裝必要的依賴：

# 基礎(chǔ)依賴
pip install langchain langchain-openai langchain-community

# LangGraph（用于高級(jí)Agent功能）
pip install langgraph

# 工具依賴
pip install tavily-python  # 搜索工具
pip install arxiv  # 學(xué)術(shù)論文搜索
pip install wikipedia-api  # Wikipedia查詢

# 可選：監(jiān)控和調(diào)試
pip install langsmith

設(shè)置API密鑰：

import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
os.environ["TAVILY_API_KEY"] = "your-tavily-key"  # 獲取地址：https://tavily.com

什么是Agent思考框架？

Agent思考框架是指導(dǎo)LLM如何系統(tǒng)化地分解問(wèn)題、推理和執(zhí)行任務(wù)的結(jié)構(gòu)化方法。這些框架讓LLM不再是簡(jiǎn)單的"問(wèn)答機(jī)器"，而是能夠像人類一樣思考、規(guī)劃和行動(dòng)的智能體。

通過(guò)規(guī)劃，LLM agents能夠推理、將復(fù)雜任務(wù)分解為更小、更可管理的部分，并為每個(gè)部分制定具體計(jì)劃。隨著任務(wù)的發(fā)展，agents還可以反思和調(diào)整他們的計(jì)劃，確保它們與現(xiàn)實(shí)世界的情況保持相關(guān)。

核心組件

一個(gè)完整的Agent框架通常包含以下核心組件：

思維（Thought）：Agent的推理過(guò)程，用于分析問(wèn)題和制定策略
行動(dòng)（Action）：Agent可以執(zhí)行的具體操作，如調(diào)用API、查詢數(shù)據(jù)庫(kù)等
觀察（Observation）：從環(huán)境中獲取的反饋信息
記憶（Memory）：存儲(chǔ)歷史信息和上下文
規(guī)劃（Planning）：制定執(zhí)行步驟和策略

二、Chain of Thought（CoT）

2.1 CoT的起源與原理

Chain of Thought prompting是一種讓模型分解多步驟問(wèn)題為中間步驟的方法。通過(guò)chain of thought prompting，足夠規(guī)模的語(yǔ)言模型（約100B參數(shù)）可以解決使用標(biāo)準(zhǔn)提示方法無(wú)法解決的復(fù)雜推理問(wèn)題。

CoT的核心思想是引導(dǎo)LLM像人類一樣，通過(guò)一系列邏輯步驟來(lái)解決問(wèn)題，而不是直接跳到答案。這種方法在2022年由Google Brain團(tuán)隊(duì)首次提出，迅速成為提升LLM推理能力的標(biāo)準(zhǔn)技術(shù)。

2.2 CoT的實(shí)現(xiàn)方式

Zero-Shot CoT

最簡(jiǎn)單的實(shí)現(xiàn)方式是在提示詞中加入"Let's think step by step"（讓我們一步步思考）：

prompt = """
問(wèn)題：如果一個(gè)班級(jí)有28個(gè)學(xué)生，其中女生比男生多4人，請(qǐng)問(wèn)男生有多少人？

讓我們一步步思考：
"""

Few-Shot CoT

通過(guò)提供示例來(lái)引導(dǎo)模型的推理過(guò)程：

prompt = """
問(wèn)題：Roger有5個(gè)網(wǎng)球。他又買了2罐網(wǎng)球，每罐有3個(gè)球。他現(xiàn)在有多少個(gè)網(wǎng)球？
思考過(guò)程：Roger開始有5個(gè)球。2罐，每罐3個(gè)球，就是2×3=6個(gè)球。5+6=11個(gè)球。
答案：11個(gè)球

問(wèn)題：食堂有23個(gè)蘋果。如果他們用了20個(gè)做午餐，又買了6個(gè)，他們有多少個(gè)蘋果？
思考過(guò)程：[模型生成]
"""

2.3 CoT的優(yōu)勢(shì)與局限

優(yōu)勢(shì)：

提高復(fù)雜推理任務(wù)的準(zhǔn)確性
使推理過(guò)程透明可解釋
適用于各種推理任務(wù)（數(shù)學(xué)、常識(shí)、符號(hào)推理等）

局限：

對(duì)較小的模型效果不佳。研究表明，CoT只在約100B參數(shù)的模型上產(chǎn)生性能提升。較小的模型會(huì)寫出不合邏輯的思維鏈，導(dǎo)致準(zhǔn)確性比標(biāo)準(zhǔn)提示更差。
仍然依賴模型內(nèi)部知識(shí)，無(wú)法獲取外部實(shí)時(shí)信息

三、ReAct

3.1 ReAct框架的創(chuàng)新

ReAct使語(yǔ)言模型能夠以交錯(cuò)的方式生成語(yǔ)言推理軌跡和文本動(dòng)作。雖然動(dòng)作會(huì)從外部環(huán)境獲得觀察反饋，但推理軌跡不會(huì)影響外部環(huán)境。相反，它們通過(guò)對(duì)上下文進(jìn)行推理并用有用的信息更新它來(lái)影響模型的內(nèi)部狀態(tài)。

ReAct框架解決了CoT的一個(gè)關(guān)鍵限制：無(wú)法與外部世界交互。它將推理（Reasoning）和行動(dòng)（Acting）結(jié)合起來(lái)，讓Agent能夠：

通過(guò)推理分解任務(wù)
執(zhí)行動(dòng)作獲取新信息
基于觀察調(diào)整策略
迭代直到解決問(wèn)題

3.2 ReAct的工作流程

典型的ReAct循環(huán)如下：

Thought 1: 我需要搜索最新的人口數(shù)據(jù)
Action 1: Search("中國(guó)人口 2024")
Observation 1: 根據(jù)最新統(tǒng)計(jì)，中國(guó)人口約為14.1億

Thought 2: 現(xiàn)在我需要計(jì)算兩倍的值
Action 2: Calculate(14.1 * 2)
Observation 2: 28.2

Thought 3: 我現(xiàn)在知道答案了
Final Answer: 中國(guó)人口的兩倍是28.2億

3.3 LangChain實(shí)現(xiàn)ReAct Agent

基礎(chǔ)實(shí)現(xiàn)

使用LangChain創(chuàng)建ReAct agent的基礎(chǔ)實(shí)現(xiàn)非常簡(jiǎn)單：

from langchain import hub
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI

# 初始化LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 定義工具
tools = [
    TavilySearchResults(max_results=3, description="搜索互聯(lián)網(wǎng)獲取最新信息")
]

# 使用預(yù)定義的ReAct提示詞模板
prompt = hub.pull("hwchase17/react")

# 創(chuàng)建ReAct agent
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 執(zhí)行查詢
result = agent_executor.invoke({
    "input": "2024年諾貝爾物理學(xué)獎(jiǎng)獲得者是誰(shuí)？他們的主要貢獻(xiàn)是什么？"
})

自定義ReAct提示詞

您可以自定義ReAct提示詞模板來(lái)控制agent的行為：

from langchain_core.prompts import PromptTemplate

template = '''Answer the following questions as best you can. 
You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: {agent_scratchpad}'''

prompt = PromptTemplate.from_template(template)

# 使用自定義提示詞創(chuàng)建agent
agent = create_react_agent(llm, tools, prompt)

3.4 LangGraph實(shí)現(xiàn)ReAct Agent

LangGraph提供了更現(xiàn)代化和功能豐富的ReAct實(shí)現(xiàn)：

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

# 創(chuàng)建帶有記憶功能的ReAct agent
memory = MemorySaver()
model = init_chat_model("anthropic:claude-3-5-sonnet-latest")
search = TavilySearch(max_results=2)
tools = [search]

# 創(chuàng)建支持多輪對(duì)話的agent
agent_executor = create_react_agent(
    model, 
    tools, 
    checkpointer=memory  # 啟用對(duì)話記憶
)

# 使用配置管理對(duì)話線程
config = {"configurable": {"thread_id": "abc123"}}

# 第一輪對(duì)話
response1 = agent_executor.invoke(
    {"messages": [{"role": "user", "content": "我叫張三，住在北京"}]},
    config
)

# 第二輪對(duì)話（agent會(huì)記住之前的信息）
response2 = agent_executor.invoke(
    {"messages": [{"role": "user", "content": "我住在哪里？"}]},
    config
)

自定義工具函數(shù)

您可以定義自定義工具來(lái)擴(kuò)展agent的能力：

from langchain_core.tools import tool

@tool
def get_weather(location: str) -> str:
    """獲取指定城市的天氣信息"""
    # 這里可以調(diào)用真實(shí)的天氣API
    if"北京"in location:
        return"北京今天晴天，溫度23°C，空氣質(zhì)量良好"
    returnf"無(wú)法獲取{location}的天氣信息"

@tool
def calculate(expression: str) -> str:
    """計(jì)算數(shù)學(xué)表達(dá)式"""
    try:
        result = eval(expression)
        returnf"計(jì)算結(jié)果：{result}"
    except:
        return"計(jì)算錯(cuò)誤，請(qǐng)檢查表達(dá)式"

# 使用自定義工具創(chuàng)建agent
tools = [get_weather, calculate, search]
agent = create_react_agent(model, tools, checkpointer=memory)

3.5 ReAct的實(shí)際應(yīng)用

ReAct框架受到人類如何使用自然語(yǔ)言進(jìn)行復(fù)雜任務(wù)的逐步規(guī)劃和執(zhí)行的啟發(fā)。ReAct agents不是實(shí)施基于規(guī)則或預(yù)定義的工作流程，而是依靠其LLM的推理能力，根據(jù)新信息或先前步驟的結(jié)果動(dòng)態(tài)調(diào)整其方法。

ReAct在以下場(chǎng)景表現(xiàn)出色：

問(wèn)答系統(tǒng)：結(jié)合Wikipedia API等外部知識(shí)源
事實(shí)驗(yàn)證：通過(guò)多次搜索驗(yàn)證信息
任務(wù)自動(dòng)化：在復(fù)雜環(huán)境中執(zhí)行多步驟任務(wù)
交互式?jīng)Q策：在游戲或模擬環(huán)境中做出決策

四、Plan-and-Execute

4.1 架構(gòu)設(shè)計(jì)理念

Plan-and-Execute架構(gòu)由兩個(gè)基本組件組成：一個(gè)規(guī)劃器（Planner），它提示LLM生成完成大型任務(wù)的多步驟計(jì)劃；執(zhí)行器（Executors），它們接受用戶查詢和計(jì)劃中的一個(gè)步驟，并調(diào)用一個(gè)或多個(gè)工具來(lái)完成該任務(wù)。

這種架構(gòu)的核心創(chuàng)新在于將規(guī)劃和執(zhí)行分離，帶來(lái)了以下優(yōu)勢(shì)：

更快的執(zhí)行速度：不需要在每個(gè)動(dòng)作后都咨詢大型LLM
成本優(yōu)化：可以使用較小的模型執(zhí)行具體任務(wù)
更好的性能：強(qiáng)制規(guī)劃器明確思考所有步驟

4.2 Plan-and-Execute vs ReAct

特性	ReAct	Plan-and-Execute
規(guī)劃方式	逐步規(guī)劃	預(yù)先制定完整計(jì)劃
LLM調(diào)用頻率	每步都需要	主要在規(guī)劃階段
靈活性	高度靈活	按計(jì)劃執(zhí)行
執(zhí)行速度	較慢	更快
成本	較高	較低
適用場(chǎng)景	探索性任務(wù)	結(jié)構(gòu)化任務(wù)

4.3 LangGraph實(shí)現(xiàn)Plan-and-Execute

方式一：使用LangChain Experimental

LangChain在experimental包中提供了Plan-and-Execute的實(shí)現(xiàn)：

# 安裝experimental包
# pip install langchain-experimental

from langchain.chains import LLMMathChain
from langchain_community.utilities import DuckDuckGoSearchAPIWrapper
from langchain_core.tools import Tool
from langchain_experimental.plan_and_execute import (
    PlanAndExecute,
    load_agent_executor,
    load_chat_planner,
)
from langchain_openai import ChatOpenAI

# 初始化LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 定義工具
search = DuckDuckGoSearchAPIWrapper()
llm_math_chain = LLMMathChain.from_llm(llm=llm)

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于搜索當(dāng)前事件和信息"
    ),
    Tool(
        name="Calculator",
        func=llm_math_chain.run,
        description="用于數(shù)學(xué)計(jì)算"
    ),
]

# 創(chuàng)建planner和executor
planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools, verbose=True)

# 創(chuàng)建Plan-and-Execute agent
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)

# 使用agent
result = agent.run("研究2024年諾貝爾物理學(xué)獎(jiǎng)獲得者并計(jì)算他們的平均年齡")

方式二：使用LangGraph自定義實(shí)現(xiàn)

包含錯(cuò)誤處理和動(dòng)態(tài)重新規(guī)劃：

from typing import Union, List, Tuple, Optional
from pydantic import BaseModel, Field
from enum import Enum

class ExecutionStatus(Enum):
    """執(zhí)行狀態(tài)枚舉"""
    SUCCESS = "success"
    FAILURE = "failure"
    PARTIAL = "partial"
    NEEDS_RETRY = "needs_retry"

class PlanEvaluation(BaseModel):
    """計(jì)劃評(píng)估結(jié)果"""
    needs_replan: bool = Field(description="是否需要重新規(guī)劃")
    reason: Optional[str] = Field(description="需要重新規(guī)劃的原因")
    new_plan: Optional[List[str]] = Field(description="新的執(zhí)行計(jì)劃")
    continue_execution: bool = Field(description="是否繼續(xù)執(zhí)行")

class SmartPlanExecute(TypedDict):
    """增強(qiáng)的狀態(tài)定義"""
    input: str
    original_plan: List[str]  # 保存原始計(jì)劃
    plan: List[str]  # 當(dāng)前待執(zhí)行的計(jì)劃
    past_steps: List[Tuple[str, str, ExecutionStatus]]  # 包含狀態(tài)
    response: Optional[str]
    replan_count: int  # 重新規(guī)劃次數(shù)
    max_replans: int  # 最大重新規(guī)劃次數(shù)

asyncdef smart_execute_step(state: SmartPlanExecute):
    """智能執(zhí)行步驟，包含錯(cuò)誤處理"""
    ifnot state["plan"]:
        return {}
    
    task = state["plan"][0]
    max_retries = 3
    
    for attempt in range(max_retries):
        try:
            # 執(zhí)行任務(wù)
            result = await agent_executor.ainvoke({
                "messages": [{"role": "user", "content": task}]
            })
            
            # 評(píng)估執(zhí)行結(jié)果
            status = await evaluate_execution_result(task, result)
            
            # 記錄執(zhí)行結(jié)果
            return {
                "past_steps": [(task, result["messages"][-1].content, status)],
                "plan": state["plan"][1:]  # 移除已執(zhí)行的步驟
            }
            
        except Exception as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # 指數(shù)退避
                continue
            else:
                # 記錄失敗
                return {
                    "past_steps": [(task, f"執(zhí)行失敗: {str(e)}", ExecutionStatus.FAILURE)],
                    "plan": state["plan"][1:]
                }

asyncdef evaluate_execution_result(task: str, result: dict) -> ExecutionStatus:
    """評(píng)估任務(wù)執(zhí)行結(jié)果"""
    evaluation_prompt = """
    評(píng)估以下任務(wù)執(zhí)行結(jié)果：
    
    任務(wù)：{task}
    結(jié)果：{result}
    
    請(qǐng)判斷執(zhí)行狀態(tài)：
    - SUCCESS: 完全成功
    - PARTIAL: 部分成功，可能需要補(bǔ)充
    - FAILURE: 執(zhí)行失敗
    - NEEDS_RETRY: 需要重試
    
    只返回狀態(tài)值。
    """
    
    response = await llm.ainvoke(
        evaluation_prompt.format(
            task=task,
            result=result["messages"][-1].content
        )
    )
    
    # 解析狀態(tài)
    status_text = response.content.strip().upper()
    for status in ExecutionStatus:
        if status.value.upper() in status_text or status.name in status_text:
            return status
    
    return ExecutionStatus.SUCCESS  # 默認(rèn)為成功

asyncdef intelligent_replan_step(state: SmartPlanExecute):
    """智能重新規(guī)劃，考慮多種因素"""
    # 檢查是否超過(guò)最大重新規(guī)劃次數(shù)
    if state["replan_count"] >= state.get("max_replans", 3):
        return {
            "response": "達(dá)到最大重新規(guī)劃次數(shù)，使用當(dāng)前結(jié)果生成報(bào)告。\n" + 
                       await generate_best_effort_report(state)
        }
    
    # 分析執(zhí)行歷史
    analysis = await analyze_execution_history(state)
    
    if analysis.needs_replan:
        # 生成新計(jì)劃
        new_plan = await generate_adaptive_plan(
            state["input"],
            state["past_steps"],
            state["plan"],
            analysis.reason
        )
        
        return {
            "plan": new_plan,
            "replan_count": state["replan_count"] + 1
        }
    
    elif len(state["plan"]) == 0:
        # 所有步驟完成，生成最終報(bào)告
        return {"response": await generate_intelligent_report(state)}
    
    else:
        # 繼續(xù)執(zhí)行
        return {}

asyncdef analyze_execution_history(state: SmartPlanExecute) -> PlanEvaluation:
    """深度分析執(zhí)行歷史"""
    # 統(tǒng)計(jì)執(zhí)行狀態(tài)
    status_counts = {}
    for _, _, status in state["past_steps"]:
        status_counts[status] = status_counts.get(status, 0) + 1
    
    # 獲取最近的執(zhí)行結(jié)果
    recent_failures = [
        (task, result) for task, result, status in state["past_steps"][-3:]
        if status in [ExecutionStatus.FAILURE, ExecutionStatus.NEEDS_RETRY]
    ]
    
    # 構(gòu)建分析提示
    analysis_prompt = """
    分析任務(wù)執(zhí)行情況并決定是否需要重新規(guī)劃：
    
    原始目標(biāo)：{input}
    執(zhí)行統(tǒng)計(jì)：{status_counts}
    最近失敗：{recent_failures}
    剩余任務(wù)：{remaining_tasks}
    
    考慮因素：
    1. 失敗率是否過(guò)高？
    2. 是否發(fā)現(xiàn)了原計(jì)劃未考慮的情況？
    3. 剩余任務(wù)是否仍然適用？
    4. 是否需要新的方法或工具？
    
    返回JSON格式：
    {{
        "needs_replan": true/false,
        "reason": "重新規(guī)劃的原因",
        "suggestions": ["建議1", "建議2"]
    }}
    """
    
    response = await llm.ainvoke(
        analysis_prompt.format(
            input=state["input"],
            status_counts=status_counts,
            recent_failures=recent_failures,
            remaining_tasks=state["plan"]
        )
    )
    
    # 解析響應(yīng)
    import json
    try:
        analysis_data = json.loads(response.content)
        return PlanEvaluation(
            needs_replan=analysis_data.get("needs_replan", False),
            reason=analysis_data.get("reason"),
            continue_execution=not analysis_data.get("needs_replan", False)
        )
    except:
        # 如果解析失敗，默認(rèn)繼續(xù)執(zhí)行
        return PlanEvaluation(
            needs_replan=False,
            continue_execution=True
        )

asyncdef generate_adaptive_plan(
    original_goal: str,
    past_steps: List[Tuple],
    remaining_tasks: List[str],
    replan_reason: str
) -> List[str]:
    """生成自適應(yīng)的新計(jì)劃"""
    adaptive_prompt = """
    基于執(zhí)行歷史生成新的計(jì)劃：
    
    原始目標(biāo)：{goal}
    已執(zhí)行步驟：{past}
    原剩余任務(wù)：{remaining}
    重新規(guī)劃原因：{reason}
    
    請(qǐng)生成新的執(zhí)行計(jì)劃，要求：
    1. 避免重復(fù)已成功的步驟
    2. 解決發(fā)現(xiàn)的問(wèn)題
    3. 使用替代方法處理失敗的任務(wù)
    4. 保持計(jì)劃簡(jiǎn)潔高效
    
    每行一個(gè)步驟：
    """
    
    response = await llm.ainvoke(
        adaptive_prompt.format(
            goal=original_goal,
            past=past_steps,
            remaining=remaining_tasks,
            reason=replan_reason
        )
    )
    
    return parse_new_plan(response.content)

# 構(gòu)建增強(qiáng)的工作流
def build_smart_plan_execute_workflow():
    workflow = StateGraph(SmartPlanExecute)
    
    # 添加節(jié)點(diǎn)
    workflow.add_node("planner", initial_plan_step)
    workflow.add_node("executor", smart_execute_step)
    workflow.add_node("analyzer", intelligent_replan_step)
    
    # 設(shè)置流程
    workflow.set_entry_point("planner")
    workflow.add_edge("planner", "executor")
    workflow.add_edge("executor", "analyzer")
    
    # 條件邊：根據(jù)分析結(jié)果決定下一步
    def route_after_analysis(state: SmartPlanExecute):
        if"response"in state and state["response"]:
            return"end"
        elif state["plan"]:
            return"executor"
        else:
            return"end"
    
    workflow.add_conditional_edges(
        "analyzer",
        route_after_analysis,
        {
            "executor": "executor",
            "end": END
        }
    )
    
    return workflow.compile()

# 使用示例
smart_agent = build_smart_plan_execute_workflow()

result = await smart_agent.ainvoke({
    "input": "分析競(jìng)爭(zhēng)對(duì)手的產(chǎn)品策略并提出改進(jìn)建議",
    "replan_count": 0,
    "max_replans": 3
})

五、框架實(shí)戰(zhàn)最佳實(shí)踐

5.1 混合使用策略

在實(shí)際項(xiàng)目中，可以根據(jù)不同階段使用不同框架：

class HybridAgent:
    def __init__(self):
        self.cot_llm = ChatOpenAI(model="gpt-3.5-turbo")
        self.react_agent = create_react_agent(...)
        self.plan_execute_agent = create_plan_and_execute_agent(...)
        
    def analyze_complexity(self, query: str) -> ComplexityAnalysis:
    """
    分析查詢的復(fù)雜度，考慮以下因素：
    - 任務(wù)是否需要多步驟推理
    - 是否需要外部信息
    - 任務(wù)的結(jié)構(gòu)化程度
    - 是否有明確的執(zhí)行路徑
    """
    pass
    
    def process(self, query):
        complexity = self.analyze_complexity(query)
        
        if complexity == "simple":
            # 簡(jiǎn)單問(wèn)題直接用CoT
            return self.cot_llm.invoke(f"{query}\n讓我們一步步思考：")
        elif complexity == "exploratory":
            # 探索性問(wèn)題用ReAct
            return self.react_agent.invoke({"input": query})
        else:
            # 復(fù)雜結(jié)構(gòu)化任務(wù)用Plan-and-Execute
            return self.plan_execute_agent.invoke({"input": query})

5.2 性能優(yōu)化技巧

緩存策略

from langchain.cache import InMemoryCache
import langchain

# 啟用緩存減少重復(fù)調(diào)用
langchain.llm_cache = InMemoryCache()

異步處理

import asyncio

async def parallel_agent_execution(queries):
    agents = [create_react_agent(llm, tools) for _ in queries]
    tasks = [agent.ainvoke({"input": q}) for agent, q in zip(agents, queries)]
    return await asyncio.gather(*tasks)

模型選擇策略

def select_model(task_type):
    model_map = {
        "reasoning": "gpt-4",  # 復(fù)雜推理用大模型
        "extraction": "gpt-3.5-turbo",  # 信息提取用快速模型
        "classification": "gpt-3.5-turbo",  # 分類任務(wù)用快速模型
        "creative": "claude-3-opus"  # 創(chuàng)意任務(wù)用Claude
    }
    return model_map.get(task_type, "gpt-3.5-turbo")

5.3 錯(cuò)誤處理和回退機(jī)制

from tenacity import retry, stop_after_attempt, wait_exponential

class RobustAgent:
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    asyncdef execute_with_retry(self, task):
        try:
            result = await self.agent.ainvoke(task)
            return result
        except Exception as e:
            # 記錄錯(cuò)誤
            logger.error(f"Agent execution failed: {e}")
            
            # 回退到更簡(jiǎn)單的方法
            if hasattr(self, 'fallback_agent'):
                returnawait self.fallback_agent.ainvoke(task)
            raise

六、最佳實(shí)踐與選擇指南

6.1 框架選擇決策樹

任務(wù)類型判斷：
├── 純推理任務(wù)（不需要外部信息）
│   └── 使用 CoT
├── 需要實(shí)時(shí)信息或工具調(diào)用
│   ├── 探索性/動(dòng)態(tài)任務(wù)
│   │   └── 使用 ReAct
│   └── 結(jié)構(gòu)化/可預(yù)測(cè)任務(wù)
│       └── 使用 Plan-and-Execute

6.2 性能優(yōu)化建議

模型選擇：

CoT需要大模型（>100B參數(shù)）才能發(fā)揮效果
ReAct和Plan-and-Execute可以混合使用大小模型

成本控制：

對(duì)于高頻任務(wù)，優(yōu)先考慮Plan-and-Execute
使用緩存機(jī)制減少重復(fù)調(diào)用

錯(cuò)誤處理：

實(shí)現(xiàn)重試機(jī)制和fallback策略
在ReAct中加入自我反思機(jī)制

6.3 實(shí)施建議

從簡(jiǎn)單開始：先用CoT驗(yàn)證基本推理能力

# 開始時(shí)使用簡(jiǎn)單的Zero-Shot CoT
prompt = "問(wèn)題：{question}\n讓我們一步步思考："

逐步升級(jí)：根據(jù)需求添加工具調(diào)用（ReAct）

# 添加基礎(chǔ)工具
tools = [search_tool]
agent = create_react_agent(llm, tools)

優(yōu)化迭代：對(duì)于成熟流程，遷移到Plan-and-Execute

# 當(dāng)流程穩(wěn)定后，使用Plan-and-Execute提升效率
planner_agent = create_plan_and_execute_agent(
    planner_llm=expensive_model,  # 只在規(guī)劃時(shí)使用大模型
    executor_llm=cheap_model,      # 執(zhí)行時(shí)使用小模型
    tools=tools
)

監(jiān)控評(píng)估：持續(xù)監(jiān)控性能和成本指標(biāo)

# 使用LangSmith進(jìn)行追蹤
from langsmith import Client
client = Client()
# 啟用追蹤來(lái)監(jiān)控agent性能

6.4 框架選擇快速參考

場(chǎng)景	推薦框架	代碼示例
簡(jiǎn)單問(wèn)答	CoT	?`?prompt = "Let's think step by step"?`?
實(shí)時(shí)信息查詢	ReAct	?`?create_react_agent(llm, [search_tool])?`?
復(fù)雜報(bào)告生成	Plan-and-Execute	?`?create_plan_and_execute_agent(llm, tools)?`?
并行任務(wù)處理	LLMCompiler	?`?parallel_execute_dag(task_graph)?`?
多輪對(duì)話	ReAct + Memory	?`?create_react_agent(llm, tools, checkpointer=memory)?`?

總結(jié)

作為開發(fā)者和技術(shù)決策者，深入理解這些框架不僅能幫助我們構(gòu)建更好的AI系統(tǒng)。無(wú)論是選擇簡(jiǎn)單直接的CoT，還是構(gòu)建復(fù)雜的多Agent系統(tǒng)，關(guān)鍵在于理解業(yè)務(wù)需求，選擇合適的工具，并持續(xù)優(yōu)化迭代。