AI Agent & LangGraph: Zero to Hero 大模型驱动的智能体 ​

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 阶段分析提示模板
stage_analyzer_inception_prompt_template = """
你是一名销售助手，帮助销售代理识别对话应该进入哪个阶段。
以下是销售对话的阶段：
1. 介绍：开始对话，介绍自己和公司。保持礼貌和尊重，同时保持对话的专业性。
2. 资格认定：确认他们是否是合适的潜在客户。
3. 价值主张：简要解释你的产品/服务如何使潜在客户受益。
4. 需求分析：询问开放性问题以发现潜在客户的需求和痛点。
5. 解决方案展示：基于潜在客户的需求，展示你的产品/服务如何解决他们的问题。
6. 异议处理：解决潜在客户可能有的任何异议。
7. 成交：询问销售问题以推进交易。

当前对话历史：
{conversation_history}

请选择应该进入的对话阶段，只需回答数字。
"""

stage_analyzer_inception_prompt = PromptTemplate(
    template=stage_analyzer_inception_prompt_template,
    input_variables=["conversation_history"],
)

llm = ChatOpenAI(temperature=0.0)
stage_analyzer_chain = LLMChain(
    llm=llm, 
    prompt=stage_analyzer_inception_prompt, 
    verbose=True
)

# 销售对话提示模板
sales_conversation_utterance_prompt_template = """
你是一名销售代表，为一家销售睡眠技术产品的公司工作。
你的名字是Ted Lasso。你遵循以下销售对话阶段的顺序：

1. 介绍：开始对话，介绍自己和公司。
2. 资格认定：确认他们是否是合适的潜在客户。
3. 价值主张：简要解释你的产品/服务如何使潜在客户受益。
4. 需求分析：询问开放性问题以发现潜在客户的需求和痛点。
5. 解决方案展示：基于潜在客户的需求，展示你的产品/服务。
6. 异议处理：解决潜在客户可能有的任何异议。
7. 成交：询问销售问题以推进交易。

当前对话阶段：{conversation_stage}
对话历史：{conversation_history}
销售代理：{salesperson_name}
公司名称：{company_name}
公司业务：{company_business}
公司价值观：{company_values}

请生成一个回复以推进销售对话。
回复应该基于之前的对话历史和当前对话阶段。
回复长度不应超过100字。
回复结尾不要说再见。

回复：
"""

sales_conversation_utterance_prompt = PromptTemplate(
    template=sales_conversation_utterance_prompt_template,
    input_variables=[
        "conversation_stage",
        "conversation_history",
        "salesperson_name",
        "company_name",
        "company_business",
        "company_values",
    ],
)

sales_conversation_utterance_chain = LLMChain(
    llm=llm, 
    prompt=sales_conversation_utterance_prompt, 
    verbose=True
)

product_catalog = """
睡眠天堂产品目录：

1. **梦境床垫**
   - 价格：$1,200
   - 特点：记忆泡沫，温度调节，10年保修
   - 适用：所有睡眠姿势，特别适合侧睡者

2. **云朵枕头**
   - 价格：$85
   - 特点：可调节高度，透气材质，防过敏
   - 适用：颈椎问题，打鼾问题

3. **智能睡眠追踪器**
   - 价格：$200
   - 特点：睡眠质量监测，智能闹钟，健康建议
   - 适用：想要改善睡眠质量的用户
"""

from langchain.vectorstores import Chroma
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.document_loaders import TextLoader

# 创建产品知识库
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_text(product_catalog)

embeddings = OpenAIEmbeddings()
docsearch = Chroma.from_texts(
    docs, embeddings, collection_name="product-knowledge-base"
)

from langchain.agents import Tool

def knowledge_base(query: str) -> str:
    """
    查询产品知识库以获取产品信息
    """
    docs = docsearch.similarity_search(query, k=2)
    docs_page_content = " ".join([d.page_content for d in docs])
    return docs_page_content

knowledge_base_tool = Tool(
    name="ProductKnowledgeBase",
    func=knowledge_base,
    description="用于回答产品相关问题的知识库。输入应该是一个搜索查询。"
)

from typing import Dict, List, Any
from langchain.chains.base import Chain
from pydantic import Field

class SalesGPT(Chain):
    """销售对话智能体的控制器"""
    
    conversation_history: List[str] = []
    current_conversation_stage: str = "1"
    stage_analyzer_chain: LLMChain = Field(...)
    sales_conversation_utterance_chain: LLMChain = Field(...)
    
    # 销售代理配置
    salesperson_name: str = "Ted Lasso"
    company_name: str = "Sleep Haven"
    company_business: str = "Sleep Haven是一家专门销售优质睡眠产品的公司。"
    company_values: str = "我们的使命是帮助人们获得更好的睡眠质量。"
    
    def retrieve_conversation_stage(self, key):
        conversation_stages = {
            "1": "介绍：开始对话，介绍自己和公司。保持礼貌和尊重，同时保持对话的专业性。你的问候应该是热情的。始终澄清你的联系原因。",
            "2": "资格认定：确认他们是否是合适的潜在客户。确保你符合他们的需求。",
            "3": "价值主张：简要解释你的产品/服务如何使潜在客户受益。使用数据来支持你的主张。",
            "4": "需求分析：询问开放性问题以发现潜在客户的需求和痛点。倾听他们的需求和痛点。",
            "5": "解决方案展示：基于潜在客户的需求，展示你的产品/服务如何解决他们的问题。",
            "6": "异议处理：解决潜在客户可能有的任何异议。准备好常见异议的回应。",
            "7": "成交：询问销售问题以推进交易。提出下一步行动。解释他们需要做什么来开始。",
        }
        return conversation_stages.get(key, "1")
    
    @property
    def input_keys(self) -> List[str]:
        return []
    
    @property
    def output_keys(self) -> List[str]:
        return []
    
    def seed_agent(self):
        # 初始化对话
        self.current_conversation_stage = self.retrieve_conversation_stage("1")
        self.conversation_history = []
    
    def determine_conversation_stage(self):
        conversation_stage_id = self.stage_analyzer_chain.run(
            conversation_history='"\n"'.join(self.conversation_history)
        )
        self.current_conversation_stage = self.retrieve_conversation_stage(conversation_stage_id)
        print(f"对话阶段: {conversation_stage_id}")
        return conversation_stage_id
    
    def human_step(self, human_input):
        # 处理人类输入
        human_input = "用户: " + human_input
        self.conversation_history.append(human_input)
    
    def step(self):
        self._call(inputs={})
    
    def _call(self, inputs: Dict[str, Any]) -> None:
        # 生成智能体的回复
        ai_message = self.sales_conversation_utterance_chain.run(
            conversation_stage=self.current_conversation_stage,
            conversation_history="\n".join(self.conversation_history),
            salesperson_name=self.salesperson_name,
            company_name=self.company_name,
            company_business=self.company_business,
            company_values=self.company_values,
        )
        
        # 添加到对话历史
        agent_name = self.salesperson_name
        ai_message = agent_name + ": " + ai_message
        self.conversation_history.append(ai_message)
        print(ai_message)
        return {}
    
    @classmethod
    def from_llm(cls, llm: ChatOpenAI, verbose: bool = False, **kwargs) -> "SalesGPT":
        """从LLM初始化SalesGPT"""
        stage_analyzer_chain = LLMChain(
            llm=llm, 
            prompt=stage_analyzer_inception_prompt, 
            verbose=verbose
        )
        sales_conversation_utterance_chain = LLMChain(
            llm=llm, 
            prompt=sales_conversation_utterance_prompt, 
            verbose=verbose
        )
        
        return cls(
            stage_analyzer_chain=stage_analyzer_chain,
            sales_conversation_utterance_chain=sales_conversation_utterance_chain,
            verbose=verbose,
            **kwargs,
        )

# 初始化销售智能体
sales_agent = SalesGPT.from_llm(llm, verbose=False)

# 开始对话
sales_agent.seed_agent()

# 手动推进对话
sales_agent.step()  # 智能体开场白

sales_agent.human_step("你好，我对你们的产品感兴趣")
sales_agent.determine_conversation_stage()
sales_agent.step()

sales_agent.human_step("我最近睡眠质量不好，经常失眠")
sales_agent.determine_conversation_stage()
sales_agent.step()

import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI

# 定义工具
tools = [TavilySearchResults(max_results=1)]

# 获取提示模板
prompt = hub.pull("hwchase17/openai-functions-agent")

# 创建LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 创建智能体
agent = create_openai_functions_agent(llm, tools, prompt)

from langgraph.prebuilt import ToolExecutor

# 创建工具执行器
tool_executor = ToolExecutor(tools)

def run_agent(data):
    """运行智能体节点"""
    agent_outcome = agent.invoke(data)
    return {"agent_outcome": agent_outcome}

def execute_tools(data):
    """执行工具节点"""
    agent_action = data['agent_outcome']
    output = tool_executor.invoke(agent_action)
    return {"intermediate_steps": [(agent_action, str(output))]}

def should_continue(data):
    """决定是否继续的条件函数"""
    if isinstance(data['agent_outcome'], AgentFinish):
        return "end"
    else:
        return "continue"

from langgraph.graph import StateGraph, END
from langchain.schema import AgentFinish

# 定义状态
from typing import TypedDict, Annotated, List, Union
from langchain.schema import AgentAction, AgentFinish, BaseMessage
import operator

class AgentState(TypedDict):
    input: str
    chat_history: list[BaseMessage]
    agent_outcome: Union[AgentAction, AgentFinish, None]
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

# 创建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)

# 设置入口点
workflow.set_entry_point("agent")

# 添加条件边
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)

# 添加普通边
workflow.add_edge('action', 'agent')

# 编译图
app = workflow.compile()

# 运行智能体
inputs = {"input": "今天天气怎么样？", "chat_history": []}
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"输出来自节点 '{key}':")
        print("---")
        print(value)
    print("\n---\n")

from langgraph.prebuilt import create_agent_executor

# 创建聊天智能体执行器
agent_executor = create_agent_executor(agent, tools)

# 运行聊天智能体
response = agent_executor.invoke({"input": "你好，我想了解一下人工智能的发展历史"})
print(response["output"])

from langchain.schema import AIMessage

def force_tool_call(data):
    """强制调用工具的节点"""
    # 检查是否直接返回了答案
    if isinstance(data['agent_outcome'], AgentFinish):
        # 强制调用搜索工具
        forced_action = AgentAction(
            tool="tavily_search_results_json",
            tool_input={"query": data['input']},
            log="强制调用搜索工具获取最新信息"
        )
        return {"agent_outcome": forced_action}
    return data

# 在图中添加强制工具调用节点
workflow.add_node("force_tool", force_tool_call)

def human_feedback(data):
    """获取人类反馈的节点"""
    print("当前智能体输出:", data['agent_outcome'].return_values['output'])
    feedback = input("请提供反馈（按回车继续，或输入修改建议）: ")
    
    if feedback.strip():
        # 如果有反馈，修改输入并重新处理
        modified_input = data['input'] + f" 请考虑以下反馈: {feedback}"
        return {"input": modified_input, "agent_outcome": None}
    
    return data

# 添加人类反馈节点
workflow.add_node("human_feedback", human_feedback)

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 加载文档
urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]

docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

# 分割文档
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

# 创建向量存储
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

from typing import List
from typing_extensions import TypedDict

class GraphState(TypedDict):
    """
    表示图状态的类型定义
    """
    question: str  # 用户问题
    generation: str  # LLM生成的答案
    documents: List[str]  # 检索到的文档

from langchain.schema import Document

def retrieve(state):
    """
    检索文档
    """
    print("---检索---")
    question = state["question"]
    
    # 检索
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents, "question": question}

def generate(state):
    """
    生成答案
    """
    print("---生成---")
    question = state["question"]
    documents = state["documents"]
    
    # RAG生成
    rag_chain = prompt | llm | StrOutputParser()
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}

def grade_documents(state):
    """
    评估文档相关性
    """
    print("---检查文档相关性---")
    question = state["question"]
    documents = state["documents"]
    
    # 评分每个文档
    filtered_docs = []
    for d in documents:
        score = retrieval_grader.invoke({"question": question, "document": d.page_content})
        grade = score['score']
        if grade == "yes":
            print("---评级：文档相关---")
            filtered_docs.append(d)
        else:
            print("---评级：文档不相关---")
            continue
    return {"documents": filtered_docs, "question": question}

# 构建图
workflow = StateGraph(GraphState)

# 定义节点
workflow.add_node("retrieve", retrieve)  # 检索
workflow.add_node("grade_documents", grade_documents)  # 评估文档
workflow.add_node("generate", generate)  # 生成

# 构建图
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "generate": "generate",
        "rewrite": "rewrite_query",
    },
)
workflow.add_edge("generate", END)

# 编译
app = workflow.compile()

# 运行
inputs = {"question": "什么是智能体记忆？"}
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"节点 '{key}':")
        pprint.pprint(value, indent=2, width=80, depth=None)
    print("\n---\n")

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field

# 网络搜索工具
web_search_tool = TavilySearchResults(k=3)

# 文档评分器
class GradeDocuments(BaseModel):
    """评估检索到的文档是否与用户问题相关"""
    
    score: str = Field(description="文档是否相关，'yes' 或 'no'")

# 创建评分链
system = """你是一个评估检索到的文档是否与用户问题相关的评分员。
如果文档包含与用户问题相关的关键词或语义，评分为 'yes'。
否则评分为 'no'。"""

grade_prompt = ChatPromptTemplate.from_messages([
    ("system", system),
    ("human", "检索到的文档: \n\n {document} \n\n 用户问题: {question}"),
])

retrieval_grader = grade_prompt | llm.with_structured_output(GradeDocuments)

def decide_to_generate(state):
    """
    决定是生成答案还是重新搜索
    """
    print("---评估评分文档---")
    
    filtered_documents = state["documents"]
    
    if not filtered_documents:
        # 所有文档都不相关，进行网络搜索
        print("---决策：所有文档都不相关，进行网络搜索---")
        return "websearch"
    else:
        # 生成答案
        print("---决策：生成答案---")
        return "generate"

def rewrite_query(state):
    """
    重写查询以改善检索效果
    """
    print("---重写查询---")
    question = state["question"]
    
    # 查询重写提示
    rewrite_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个查询重写助手。重写用户查询以改善检索效果。"),
        ("human", "原始查询: {question} \n 重写查询:"),
    ])
    
    question_rewriter = rewrite_prompt | llm | StrOutputParser()
    better_question = question_rewriter.invoke({"question": question})
    
    return {"question": better_question}

def web_search(state):
    """
    网络搜索
    """
    print("---网络搜索---")
    question = state["question"]
    
    # 网络搜索
    docs = web_search_tool.invoke({"query": question})
    web_results = "\n".join([d["content"] for d in docs])
    web_results = Document(page_content=web_results)
    
    return {"documents": [web_results], "question": question}

# 构建图
workflow = StateGraph(GraphState)

# 定义节点
workflow.add_node("retrieve", retrieve)  # 检索
workflow.add_node("grade_documents", grade_documents)  # 评估文档
workflow.add_node("generate", generate)  # 生成
workflow.add_node("rewrite_query", rewrite_query)  # 重写查询
workflow.add_node("websearch", web_search)  # 网络搜索

# 构建图
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "websearch": "websearch",
        "generate": "generate",
    },
)
workflow.add_edge("websearch", "generate")
workflow.add_edge("generate", END)

# 编译
app = workflow.compile()

# 测试相关查询
inputs = {"question": "智能体规划是什么？"}
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"节点 '{key}':")
        print(value)
    print("\n---\n")

# 测试不相关查询
inputs = {"question": "今天的股票市场表现如何？"}
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"节点 '{key}':")
        print(value)
    print("\n---\n")

import functools
import operator
from typing import Annotated, Any, Dict, List, Optional, Sequence, TypedDict
import uuid

from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import create_agent_executor

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool

tavily_tool = TavilySearchResults(max_results=5)

@tool
def scrape_webpages(urls: List[str]) -> str:
    """使用requests和beautifulsoup抓取网页内容"""
    # 实现网页抓取逻辑
    return "抓取的网页内容..."

@tool  
def create_outline(points: List[str]) -> str:
    """创建内容大纲"""
    outline = "\n".join([f"{i+1}. {point}" for i, point in enumerate(points)])
    return f"内容大纲:\n{outline}"

@tool
def read_document(document_path: str) -> str:
    """读取文档内容"""
    # 实现文档读取逻辑
    return "文档内容..."

@tool
def write_document(content: str, file_path: str) -> str:
    """写入文档"""
    # 实现文档写入逻辑
    return f"文档已保存到 {file_path}"

def create_agent(llm: ChatOpenAI, tools: list, system_prompt: str):
    """创建一个智能体"""
    from langchain.agents import create_openai_functions_agent
    from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="messages"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])
    
    agent = create_openai_functions_agent(llm, tools, prompt)
    executor = create_agent_executor(agent, tools)
    return executor

def agent_node(state, agent, name):
    """智能体节点包装器"""
    result = agent.invoke(state)
    return {"messages": [HumanMessage(content=result["output"], name=name)]}

# 创建研究智能体
research_agent = create_agent(
    llm,
    [tavily_tool, scrape_webpages],
    "你是一个研究助手。负责搜索和收集信息。"
    "你应该提供准确、全面的研究结果。"
)

research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")

# 创建图表生成智能体  
chart_agent = create_agent(
    llm,
    [create_outline, read_document],
    "你是一个图表生成专家。负责创建数据可视化图表。"
    "你应该根据数据创建清晰、有意义的图表。"
)

chart_node = functools.partial(agent_node, agent=chart_agent, name="ChartGenerator")

# 研究团队路由器
def research_router(state):
    messages = state["messages"]
    last_message = messages[-1]
    if "图表" in last_message.content or "可视化" in last_message.content:
        return "chart_generator"
    return "researcher"

# 构建研究团队图
research_graph = StateGraph(AgentState)
research_graph.add_node("researcher", research_node)
research_graph.add_node("chart_generator", chart_node)
research_graph.add_conditional_edges(
    "researcher",
    research_router,
    {"chart_generator": "chart_generator", "researcher": END}
)
research_graph.add_edge("chart_generator", END)
research_graph.set_entry_point("researcher")

research_chain = research_graph.compile()

# 创建文档撰写智能体
doc_writer_agent = create_agent(
    llm,
    [write_document, read_document],
    "你是一个文档撰写专家。负责撰写高质量的文档。"
    "你应该确保文档结构清晰、内容准确。"
)

doc_writer_node = functools.partial(agent_node, agent=doc_writer_agent, name="DocWriter")

# 创建编辑智能体
editor_agent = create_agent(
    llm,
    [read_document, write_document],
    "你是一个编辑。负责审查和改进文档质量。"
    "你应该检查语法、逻辑和整体质量。"
)

editor_node = functools.partial(agent_node, agent=editor_agent, name="Editor")

# 文档团队路由器
def doc_router(state):
    messages = state["messages"]
    last_message = messages[-1]
    if "编辑" in last_message.content or "审查" in last_message.content:
        return "editor"
    return "doc_writer"

# 构建文档团队图
authoring_graph = StateGraph(AgentState)
authoring_graph.add_node("doc_writer", doc_writer_node)
authoring_graph.add_node("editor", editor_node)
authoring_graph.add_conditional_edges(
    "doc_writer",
    doc_router,
    {"editor": "editor", "doc_writer": END}
)
authoring_graph.add_edge("editor", END)
authoring_graph.set_entry_point("doc_writer")

authoring_chain = authoring_graph.compile()

# 定义顶层状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next: str

def supervisor_agent(state):
    """监督智能体决定下一步行动"""
    messages = state["messages"]
    
    # 简单的路由逻辑
    last_message = messages[-1].content.lower()
    
    if any(keyword in last_message for keyword in ["研究", "搜索", "调查"]):
        return {"next": "research_team"}
    elif any(keyword in last_message for keyword in ["写作", "文档", "报告"]):
        return {"next": "authoring_team"}
    else:
        return {"next": "FINISH"}

# 构建顶层图
top_graph = StateGraph(AgentState)
top_graph.add_node("supervisor", supervisor_agent)
top_graph.add_node("research_team", research_chain)
top_graph.add_node("authoring_team", authoring_chain)

top_graph.set_entry_point("supervisor")
top_graph.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {
        "research_team": "research_team",
        "authoring_team": "authoring_team",
        "FINISH": END
    }
)
top_graph.add_edge("research_team", "supervisor")
top_graph.add_edge("authoring_team", "supervisor")

hierarchical_chain = top_graph.compile()

# 运行示例
response = hierarchical_chain.invoke({
    "messages": [HumanMessage(content="请研究人工智能的最新发展趋势")]
})
print(response)

from typing import List, Tuple
from langchain_core.pydantic_v1 import BaseModel, Field

class Plan(BaseModel):
    """执行计划"""
    steps: List[str] = Field(description="执行步骤列表")

class Response(BaseModel):
    """执行响应"""
    response: str = Field(description="执行结果")

# 规划器节点
def planner_node(state):
    messages = state["messages"]
    
    planner_prompt = ChatPromptTemplate.from_template(
        "给定以下目标，创建一个详细的执行计划。将复杂任务分解为简单的步骤。\n"
        "目标: {goal}\n"
        "计划:"
    )
    
    planner = planner_prompt | llm.with_structured_output(Plan)
    plan = planner.invoke({"goal": messages[0].content})
    
    return {"plan": plan.steps}

# 执行器节点
def executor_node(state):
    plan = state["plan"]
    results = []
    
    for step in plan:
        # 执行每个步骤
        executor_prompt = ChatPromptTemplate.from_template(
            "执行以下步骤: {step}\n"
            "之前的结果: {previous_results}\n"
            "执行结果:"
        )
        
        executor = executor_prompt | llm.with_structured_output(Response)
        result = executor.invoke({
            "step": step,
            "previous_results": "\n".join(results)
        })
        
        results.append(result.response)
    
    return {"results": results}

# 再规划器节点
def replanner_node(state):
    original_plan = state["plan"]
    results = state["results"]
    
    replanner_prompt = ChatPromptTemplate.from_template(
        "原始计划: {original_plan}\n"
        "执行结果: {results}\n"
        "是否需要调整计划？如果需要，提供新的计划:"
    )
    
    replanner = replanner_prompt | llm.with_structured_output(Plan)
    new_plan = replanner.invoke({
        "original_plan": original_plan,
        "results": results
    })
    
    return {"plan": new_plan.steps}

class Step(BaseModel):
    """执行步骤"""
    id: int = Field(description="步骤ID")
    tool: str = Field(description="使用的工具")
    tool_input: str = Field(description="工具输入，可以引用之前的结果如#E1")

class Plan(BaseModel):
    """ReWOO计划"""
    steps: List[Step] = Field(description="执行步骤")

# ReWOO规划器
def rewoo_planner(state):
    query = state["query"]
    
    planner_prompt = ChatPromptTemplate.from_template(
        "为以下查询创建一个执行计划。使用#E1, #E2等引用之前步骤的结果。\n"
        "可用工具: search, calculator, python\n"
        "查询: {query}\n"
        "计划:"
    )
    
    planner = planner_prompt | llm.with_structured_output(Plan)
    plan = planner.invoke({"query": query})
    
    return {"plan": plan}

# ReWOO执行器
def rewoo_worker(state):
    plan = state["plan"]
    results = {}
    
    for step in plan.steps:
        # 替换变量引用
        tool_input = step.tool_input
        for var, value in results.items():
            tool_input = tool_input.replace(f"#{var}", str(value))
        
        # 执行工具
        if step.tool == "search":
            result = tavily_tool.invoke({"query": tool_input})
        elif step.tool == "calculator":
            result = eval(tool_input)  # 简化示例
        else:
            result = f"执行 {step.tool} 工具的结果"
        
        results[f"E{step.id}"] = result
    
    return {"results": results}

# ReWOO求解器
def rewoo_solver(state):
    query = state["query"]
    results = state["results"]
    
    solver_prompt = ChatPromptTemplate.from_template(
        "基于以下执行结果回答查询:\n"
        "查询: {query}\n"
        "执行结果: {results}\n"
        "答案:"
    )
    
    solver = solver_prompt | llm | StrOutputParser()
    answer = solver.invoke({"query": query, "results": results})
    
    return {"answer": answer}

import asyncio
from typing import Dict, Set

class Task(BaseModel):
    """任务定义"""
    id: str = Field(description="任务ID")
    tool: str = Field(description="使用的工具")
    args: Dict[str, Any] = Field(description="工具参数")
    dependencies: List[str] = Field(default=[], description="依赖的任务ID")

class TaskDAG(BaseModel):
    """任务DAG"""
    tasks: List[Task] = Field(description="任务列表")

# LLMCompiler规划器
def llm_compiler_planner(state):
    query = state["query"]
    
    planner_prompt = ChatPromptTemplate.from_template(
        "将以下查询分解为可并行执行的任务DAG。\n"
        "查询: {query}\n"
        "任务DAG:"
    )
    
    planner = planner_prompt | llm.with_structured_output(TaskDAG)
    dag = planner.invoke({"query": query})
    
    return {"dag": dag}

# 任务获取单元
def task_fetcher(state):
    dag = state["dag"]
    completed = set(state.get("completed", []))
    
    # 找到可以执行的任务（所有依赖都已完成）
    ready_tasks = []
    for task in dag.tasks:
        if task.id not in completed:
            if all(dep in completed for dep in task.dependencies):
                ready_tasks.append(task)
    
    return {"ready_tasks": ready_tasks}

# 并行执行器
async def parallel_executor(state):
    ready_tasks = state["ready_tasks"]
    results = state.get("results", {})
    
    # 并行执行所有就绪任务
    async def execute_task(task):
        # 替换参数中的依赖引用
        args = task.args.copy()
        for key, value in args.items():
            if isinstance(value, str) and value.startswith("$"):
                dep_id = value[1:]  # 移除$前缀
                args[key] = results.get(dep_id, value)
        
        # 执行工具
        if task.tool == "search":
            result = tavily_tool.invoke(args)
        else:
            result = f"执行任务 {task.id} 的结果"
        
        return task.id, result
    
    # 并行执行
    task_results = await asyncio.gather(*[execute_task(task) for task in ready_tasks])
    
    # 更新结果
    for task_id, result in task_results:
        results[task_id] = result
    
    completed = set(state.get("completed", []))
    completed.update([task.id for task in ready_tasks])
    
    return {"results": results, "completed": list(completed)}

from langchain_core.pydantic_v1 import BaseModel, Field

class RetrievalDecision(BaseModel):
    """检索决策"""
    should_retrieve: bool = Field(description="是否需要检索")
    reasoning: str = Field(description="决策理由")

class RelevanceScore(BaseModel):
    """相关性评分"""
    score: float = Field(description="相关性分数 0-1")
    reasoning: str = Field(description="评分理由")

class SupportScore(BaseModel):
    """支持度评分"""
    score: float = Field(description="支持度分数 0-1")
    reasoning: str = Field(description="评分理由")

# Self-RAG节点定义
def retrieval_decision_node(state):
    """决定是否需要检索"""
    query = state["query"]
    
    decision_prompt = ChatPromptTemplate.from_template(
        "判断以下查询是否需要检索外部信息:\n"
        "查询: {query}\n"
        "决策:"
    )
    
    decision_maker = decision_prompt | llm.with_structured_output(RetrievalDecision)
    decision = decision_maker.invoke({"query": query})
    
    return {"retrieval_decision": decision}

def relevance_evaluation_node(state):
    """评估检索内容的相关性"""
    query = state["query"]
    documents = state["documents"]
    
    relevance_scores = []
    for doc in documents:
        relevance_prompt = ChatPromptTemplate.from_template(
            "评估以下文档与查询的相关性:\n"
            "查询: {query}\n"
            "文档: {document}\n"
            "相关性评分:"
        )
        
        evaluator = relevance_prompt | llm.with_structured_output(RelevanceScore)
        score = evaluator.invoke({"query": query, "document": doc.page_content})
        relevance_scores.append(score)
    
    return {"relevance_scores": relevance_scores}

def generate_with_reflection(state):
    """生成答案并进行自我反思"""
    query = state["query"]
    documents = state.get("documents", [])
    
    # 生成初始答案
    generation_prompt = ChatPromptTemplate.from_template(
        "基于以下信息回答查询:\n"
        "查询: {query}\n"
        "参考信息: {documents}\n"
        "答案:"
    )
    
    generator = generation_prompt | llm | StrOutputParser()
    answer = generator.invoke({"query": query, "documents": documents})
    
    # 自我反思
    reflection_prompt = ChatPromptTemplate.from_template(
        "评估以下答案的质量:\n"
        "查询: {query}\n"
        "答案: {answer}\n"
        "参考信息: {documents}\n"
        "评估:"
    )
    
    reflector = reflection_prompt | llm.with_structured_output(SupportScore)
    reflection = reflector.invoke({
        "query": query, 
        "answer": answer, 
        "documents": documents
    })
    
    return {"answer": answer, "reflection": reflection}

from typing import List, Dict, Any

class ReflexionState(TypedDict):
    """Reflexion状态"""
    task: str
    attempt: int
    max_attempts: int
    trajectory: List[str]
    reflection_history: List[str]
    success: bool
    final_answer: str

def execute_task(state):
    """执行任务"""
    task = state["task"]
    attempt = state["attempt"]
    reflection_history = state.get("reflection_history", [])
    
    # 构建提示，包含历史反思
    reflection_context = "\n".join([
        f"反思 {i+1}: {reflection}" 
        for i, reflection in enumerate(reflection_history)
    ])
    
    execution_prompt = ChatPromptTemplate.from_template(
        "执行以下任务 (尝试 {attempt}):\n"
        "任务: {task}\n"
        "历史反思:\n{reflection_context}\n"
        "基于历史反思，请改进你的方法。\n"
        "执行过程:"
    )
    
    executor = execution_prompt | llm | StrOutputParser()
    trajectory = executor.invoke({
        "task": task,
        "attempt": attempt,
        "reflection_context": reflection_context
    })
    
    return {"trajectory": [trajectory]}

def evaluate_execution(state):
    """评估执行结果"""
    task = state["task"]
    trajectory = state["trajectory"]
    
    evaluation_prompt = ChatPromptTemplate.from_template(
        "评估以下任务执行是否成功:\n"
        "任务: {task}\n"
        "执行轨迹: {trajectory}\n"
        "成功标准: 任务是否完成，答案是否正确\n"
        "评估结果 (成功/失败):"
    )
    
    evaluator = evaluation_prompt | llm | StrOutputParser()
    evaluation = evaluator.invoke({
        "task": task,
        "trajectory": "\n".join(trajectory)
    })
    
    success = "成功" in evaluation.lower()
    return {"success": success, "evaluation": evaluation}

def generate_reflection(state):
    """生成反思"""
    if state["success"]:
        return state  # 成功则不需要反思
    
    task = state["task"]
    trajectory = state["trajectory"]
    evaluation = state["evaluation"]
    
    reflection_prompt = ChatPromptTemplate.from_template(
        "分析以下失败的任务执行，生成改进建议:\n"
        "任务: {task}\n"
        "执行轨迹: {trajectory}\n"
        "评估结果: {evaluation}\n"
        "请分析失败原因并提供具体的改进建议:\n"
        "反思:"
    )
    
    reflector = reflection_prompt | llm | StrOutputParser()
    reflection = reflector.invoke({
        "task": task,
        "trajectory": "\n".join(trajectory),
        "evaluation": evaluation
    })
    
    reflection_history = state.get("reflection_history", [])
    reflection_history.append(reflection)
    
    return {"reflection_history": reflection_history}

def should_continue(state):
    """决定是否继续尝试"""
    if state["success"]:
        return "end"
    elif state["attempt"] >= state["max_attempts"]:
        return "end"
    else:
        return "continue"

# 构建Reflexion图
reflexion_graph = StateGraph(ReflexionState)

reflexion_graph.add_node("execute", execute_task)
reflexion_graph.add_node("evaluate", evaluate_execution)
reflexion_graph.add_node("reflect", generate_reflection)

reflexion_graph.set_entry_point("execute")
reflexion_graph.add_edge("execute", "evaluate")
reflexion_graph.add_conditional_edges(
    "evaluate",
    should_continue,
    {
        "continue": "reflect",
        "end": END
    }
)
reflexion_graph.add_edge("reflect", "execute")

reflexion_app = reflexion_graph.compile()

# 运行示例
result = reflexion_app.invoke({
    "task": "解决数学问题：如果一个圆的半径是5，求其面积",
    "attempt": 1,
    "max_attempts": 3,
    "trajectory": [],
    "reflection_history": [],
    "success": False,
    "final_answer": ""
})

print("最终结果:", result)```

输出示例：