# 基礎(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

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

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

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

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ò)程：[模型生成]
"""

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億

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)是什么？"
})

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)

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
)

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)

# 安裝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ì)算他們的平均年齡")

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
})

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})

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")

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

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

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

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

# 當(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
)

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