代码（需要修改调试）

父类

import json
from dataclasses import dataclass, field
from typing import Annotated, Sequence, Literal, Dict, List, TypeAlias, Any, TypedDict
from typing_extensions import TypedDict

from langchain_core.tools import tool
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langgraph.managed import IsLastStep

AnyMessage: TypeAlias = BaseMessage


@dataclass
class InputState:
    """输入状态类"""
    messages: Annotated[Sequence[AnyMessage], add_messages] = field(
        default_factory=list
    )

@dataclass
class State(InputState):
    """扩展的状态类"""
    is_last_step: IsLastStep = field(default=False)
class BaseAgent:
    """Agent 基础类，封装了所有公共方法和属性"""



    # 默认的 ReAct 提示模板
    REACT_PROMPT = '''  
    你被设计用于帮助完成各种输入任务，包括回答问题、内容创作、自动化处理等。  

    ## 工具  
    你可以使用各种工具，并且需要自行决定使用工具的顺序，以完成当前任务。  
    这可能需要将任务拆分为多个子任务，并使用不同的工具来完成各个子任务。  
    你可以使用以下工具：  

    {tools_desc}  

    ## 输出格式  
    如果本次需要使用工具完成某个子任务，请按照以下格式输出：  

    Thought: 我需要使用一个工具来帮助回答这个问题。  
    Action: 工具名称 (从 {tool_names} 中选择一个工具)，如果需要使用工具的话。  
    Action Input: 传递给工具的输入，使用 JSON 格式表示参数（例如：{{"query": "你好"}}）。  

    注意：  
    * 始终以“Thought”开头。  
    * 绝对不要在你的响应外部使用 Markdown 代码标记，但如果在你的回答中需要，你可以在适当的位置使用代码标记。  
    * 请使用有效的JSON 格式作为Action Input。不要使用类似 {{'input': '你好世界', 'num_beams': 5}} 这种错误格式。  
    * 一次响应最多只能使用一个工具以完成一个子任务。不要在一次响应中出现多个Action。  

    如果你已经获得足够的信息而来输出最终回答，则必须使用以下两种格式之一来输出最后答案:  

    Thought: 我可以在不使用更多工具的情况下回答问题。  
    Answer: [你的回答 (与用户问题的语言相同)]  

    Thought: 我无法使用提供的工具回答该问题。  
    Answer: [你的回答 (与用户问题的语言相同)]  

    注意：  
    * 确保所有子任务都完成后才输出最终答案  
    * 最终回答尽量涵盖用户任务的所有方面  

    ## 当前对话  
    以下是当前的对话历史, 由人类、AI的消息交替组成。  
    '''



    def __init__(self):
        """初始化方法"""
        self.llm = None  # 子类需要设置具体的LLM
        self.tools = []  # 子类需要设置具体的工具
        self.graph = None  # 工作流图


    async def call_model(self, state: State, config: RunnableConfig) -> Dict[str, List[AIMessage]]:
        """调用模型的核心方法"""
        # 生成工具描述
        tools_desc = "\n".join([
            f"- {tool.name}: {tool.description}\n"
            for tool in self.tools
        ])

        tool_names = [tool.name for tool in self.tools]

        # 生成 ReAct 提示
        system_prompt = self.REACT_PROMPT.format(
            tools_desc=tools_desc,
            tool_names=tool_names
        )

        # 预处理消息
        state.messages = self.preprocess_messages(state.messages)

        # 调用 LLM
        response = await self.llm.ainvoke([SystemMessage(content=system_prompt)] + state.messages)
        content = response.content

        print("\n===========================Reasoning================================")
        print(f'{content}')
        print("=========================Reasoning End================================\n")

        # 处理工具调用
        if "Action:" in content and "Action Input:" in content:
            action_lines = [line for line in content.split('\n') if
                            line.startswith('Action:') or line.startswith('Action Input:')]
            tool_name = action_lines[0].replace('Action:', '').strip()
            tool_input = action_lines[1].replace('Action Input:', '').replace("'", '"').strip()
            tool_input = tool_input.replace('\\', '\\\\')

            response.tool_calls = [{
                "id": "call_1",
                "type": "function",
                "name": tool_name,
                "args": json.loads(tool_input)
            }]

        # 处理最后一步
        if state.is_last_step and response.tool_calls:
            return {
                "messages": [
                    AIMessage(
                        content="对不起，我在指定的步骤数内无法找到问题的答案。"
                    )
                ]
            }

        return {"messages": [response]}

    def route_model_output(self, state: State) -> Literal["__end__", "tools"]:
        """路由模型输出决定下一步"""
        last_message = state.messages[-1]
        if not isinstance(last_message, AIMessage):
            raise ValueError(
                f"Expected AIMessage in output edges, but got {type(last_message).__name__}"
            )

        if not last_message.tool_calls:
            return "__end__"

        return "tools"

    def preprocess_messages(self, messages):
        """预处理消息，子类可以重写此方法"""
        return messages

    async def run(self, input_messages):
        """运行Agent的入口方法"""
        if not self.graph:
            self.initialize_graph()
        return await self.graph.ainvoke({"messages": input_messages})

    async def main(self, input_text):
        """主方法，方便测试"""
        messages = [HumanMessage(content=input_text)]
        result = await self.run(messages)
        # print(f"final answer : {result[-1].content}")
        return result

基类

import json

from agents.BaseAgent import BaseAgent, State, InputState
from langchain_ollama import ChatOllama
from langchain_core.messages import ToolMessage, HumanMessage, SystemMessage, AIMessage
from langchain_core.tools import tool
from typing import List, Dict, Any

import json
from dataclasses import dataclass, field
from typing import Annotated, Sequence, Literal, Dict, List, TypeAlias, Any, TypedDict
from typing_extensions import TypedDict

from langchain_core.tools import tool
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langgraph.managed import IsLastStep

from langchain_core.messages import ToolMessage

AnyMessage: TypeAlias = BaseMessage
class ReActAgent(BaseAgent):
    """ReAct智能体实现，继承自BaseAgent"""

    # 覆盖父类的REACT_PROMPT
    REACT_PROMPT = '''  
        你被设计用于帮助完成各种输入任务，包括回答问题、内容创作、自动化处理等。  

        ## 工具  
        你可以使用各种工具，并且需要自行决定使用工具的顺序，以完成当前任务。  
        这可能需要将任务拆分为多个子任务，并使用不同的工具来完成各个子任务。  
        你可以使用以下工具：  

        {tools_desc}  

        ## 输出格式  
        如果本次需要使用工具完成某个子任务，请按照以下格式输出：  

        Thought: 我需要使用一个工具来帮助回答这个问题。  
        Action: 工具名称 (从 {tool_names} 中选择一个工具)，如果需要使用工具的话。  
        Action Input: 传递给工具的输入，使用 JSON 格式表示参数（例如：{{"query": "你好"}}）。  

        注意：  
        * 一次响应最多只能使用一个工具以完成一个子任务。不要在一次响应中出现多个Action。
        * 始终以“Thought”开头。  
        * 绝对不要在你的响应外部使用 Markdown 代码标记，但如果在你的回答中需要，你可以在适当的位置使用代码标记。  
        * 请使用有效的JSON 格式作为Action Input。不要使用类似 {{'input': '你好世界', 'num_beams': 5}} 这种错误格式。  
        * 一次响应最多只能使用一个工具以完成一个子任务。不要在一次响应中出现多个Action。  

        如果你已经获得足够的信息而来输出最终回答，则必须使用以下两种格式之一来输出最后答案:  

        Thought: 我可以在不使用更多工具的情况下回答问题。  
        Answer: [你的回答 (与用户问题的语言相同)]  

        Thought: 我无法使用提供的工具回答该问题。  
        Answer: [你的回答 (与用户问题的语言相同)]  

        注意：  
        * 确保所有子任务都完成后才输出最终答案  
        * 最终回答尽量涵盖用户任务的所有方面  

        ## 当前对话  
        以下是当前的对话历史, 由人类、AI的消息交替组成。  
        '''



    def __init__(self):
        """初始化ReActAgent"""
        super().__init__()
        # 设置Ollama模型
        self.llm = ChatOllama(
            model="qwen2:7b",
            temperature=0.7,
            repeat_penalty=1.2,
            top_k=50,
            top_p=0.9,
            base_url="http://localhost:11434"
        )
        # 设置工具集
        self.tools = [
            self.network_search,
            self.query_database,
            self.document_summarizer,
            self.email
        ]
        # 初始化工作流图
        self.initialize_graph()

    def initialize_graph(self):
        """初始化工作流图"""
        builder = StateGraph(State, input=InputState)
        builder.add_node("agent", self.call_model)
        builder.add_node("tools", ToolNode(self.tools))
        builder.add_edge("__start__", "agent")

        # 添加条件边
        builder.add_conditional_edges(
            "agent",
            self.route_model_output,
        )
        builder.add_edge("tools", "agent")

        # 编译图
        self.graph = builder.compile()
        self.graph.name = self.__class__.__name__
        return self.graph


    async def call_model(self, state: State, config: RunnableConfig) -> Dict[str, List[AIMessage]]:
        """调用模型的核心方法"""
        # 生成工具描述
        tools_desc = "\n".join([
            f"- {tool.name}: {tool.description}\n"
            for tool in self.tools
        ])

        tool_names = [tool.name for tool in self.tools]

        # 生成 ReAct 提示
        system_prompt = self.REACT_PROMPT.format(
            tools_desc=tools_desc,
            tool_names=tool_names
        )

        # 预处理消息
        state.messages = self.preprocess_messages(state.messages)
        # state.messages = [HumanMessage(content=msg.content) if "ToolMessage" in msg.__class__.__name__ else msg for msg
        #                   in
        #                   state.messages]
        # 调用 LLM
        response = await self.llm.ainvoke([SystemMessage(content=system_prompt)] + state.messages)
        content = response.content

        print("\n===========================Reasoning================================")
        print(f'{content}')
        print("=========================Reasoning End================================\n")

        if "Action:" in content and "Action Input:" in content:
            action_lines = [line for line in content.split('\n') if
                            line.startswith('Action:') or line.startswith('Action Input:')]
            tool_name = action_lines[0].replace('Action:', '').strip()
            tool_input = action_lines[1].replace('Action Input:', '').strip()

            # 修复1：移除Python格式字符串标记
            tool_input = tool_input.replace("f'", "'").replace('f"', '"')

            # 修复2：处理中文特殊字符
            tool_input = tool_input.replace('《', '\\"').replace('》', '\\"')

            # 修复3：单引号转双引号
            tool_input = tool_input.replace("'", '"')

            # 修复4：转义特殊字符
            tool_input = tool_input.replace('\n', '\\n').replace('\t', '\\t')

            try:
                # 修复5：安全解析
                tool_args = json.loads(tool_input)
            except json.JSONDecodeError:
                # 修复6：回退方案 - 创建基本结构
                tool_args = {"input": tool_input}

            response.tool_calls = [{
                "id": "call_" + str(len(state.messages) + 1),
                "type": "function",
                "name": tool_name,
                "args": tool_args
            }]
        else:
            print(f"未解析到工具")

        # 处理最后一步
        if state.is_last_step and response.tool_calls:
            return {
                "messages": [
                    AIMessage(
                        content="对不起，我在指定的步骤数内无法找到问题的答案。"
                    )
                ]
            }

        return {"messages": [response]}

    def route_model_output(self, state: State) -> Literal["__end__", "tools"]:
        """路由模型输出决定下一步"""
        last_message = state.messages[-1]
        if not isinstance(last_message, AIMessage):
            raise ValueError(
                f"Expected AIMessage in output edges, but got {type(last_message).__name__}"
            )

        if not last_message.tool_calls:
            return "__end__"

        return "tools"

    def initialize_graph(self):
        """初始化工作流图"""
        builder = StateGraph(State, input=InputState)
        builder.add_node("agent", self.call_model)
        builder.add_node("tools", ToolNode(self.tools))
        builder.add_edge("__start__", "agent")

        # 添加条件边
        builder.add_conditional_edges(
            "agent",
            self.route_model_output,
        )
        builder.add_edge("tools", "agent")

        # 编译图
        self.graph = builder.compile()
        self.graph.name = self.__class__.__name__
        return self.graph


    def preprocess_messages(self, messages):
        """重写预处理方法，转换ToolMessage为HumanMessage"""
        return [
            HumanMessage(content=msg.content)
            if isinstance(msg, ToolMessage)
            else msg
            for msg in messages
        ]

    # 定义4个模拟工具
    @tool
    def network_search(query: str) -> str:
        """
        用于执行网络搜索并返回搜索结果
        Args:
            query (str): 要进行网络搜索的关键词
        """
        print(f'调用tool: network_search, 输入: {query}', 'green')
        try:
            # 调用 TavilySearchResults 来执行搜索，模拟返回结果
            # results = TavilySearchResults(max_results=2).invoke({"query": query})
            # results = "哪吒票房80亿，成本30亿，上映日期2025年10月1日！"
            results = "哪吒票房80亿，成本30亿"
            return f"搜索结果: {results}"
        except Exception as e:
            return f"搜索错误: {str(e)}"

    @tool
    def query_database(query: str) -> str:
        """
        用来查询公司数据库获取电影的投放广告费用，Args只能包含字符串
        Args:
            query (str): 要查询的电影名称，比如：哪吒2
        """
        print(f'调用tool: query_database, 输入: {query}', 'green')
        return "300万"

    @tool
    def document_summarizer(text: str) -> str:
        """
        用来提炼与总结文本的核心内容，形成摘要
        Args:
            text (str): 要总结的文本内容
        """
        print(f'调用tool: document_summarizer, 输入: {text}', 'green')
        return "总结结果：完美"

    @tool
    def email(recipient, subject, body) -> str:
        """
        用于发送电子邮件
        Args:
            recipient (str): 收件人的电子邮件地址
            subject (str): 电子邮件的主题
            body (str): 电子邮件的正文
        """
        print(f'调用tool: email, 输入: {recipient}', 'green')
        return "邮件已经发送给小明"


# 测试代码
# async def main():
#     agent = ReActAgent()
#     result = await agent.main("搜索《[哪吒2](@replace=10001)》的最新票房，并计算利润。最后结果发邮件到test@gmail.com")
#     print(f"最终结果: {result}")
#
#
# if __name__ == "__main__":
#     import asyncio
#
#     asyncio.run(main())


async def main():
    # 输入消息
    # input_message = "搜索《哪吒2》的最新票房，并计算利润。最后结果发邮件到test@gmail.com"
    # messages = [HumanMessage(content=input_message)]

    agent = ReActAgent()
    result = await agent.main("搜索《哪吒2》的最新票房，然后查询数据库获取我们公司给哪吒2投放了多少钱的广告，然后将票房和广告费汇总结果发邮件到test@gmail.com")
    # print(f"最终结果: {result}")


    print("\n=== 工作流执行完成 ===")

    # 1. 查找并显示最终答案
    final_answer = None
    for message in reversed(result["messages"]):
        if isinstance(message, AIMessage):
            # 查找明确标记为"Answer"的内容
            if "Answer:" in message.content:
                # 提取Answer之后的内容
                final_answer = message.content.split("Answer:", 1)[-1].strip()
                break
            elif not message.tool_calls:  # 最后一条非工具调用的AI消息
                final_answer = message.content
                break

    if final_answer:
        print(f"最终结果: {final_answer}")
    else:
        print("未能找到明确的最终答案")

    # 2. 检查工具调用结果
    print("\n工具调用历史:")
    for message in result["messages"]:
        if isinstance(message, ToolMessage):
            # 打印工具调用返回的结果
            print(f"工具 '{message.tool_call_id}': {message.content}")

    # 3. 检查邮件发送状态（特殊处理）
    email_sent = False
    for message in result["messages"]:
        if isinstance(message, ToolMessage) and "邮件已" in message.content:
            email_sent = True
            # 提取收件人信息（假设邮件工具返回内容中有邮箱）
            if "@" in message.content:
                email_parts = message.content.split("邮件已发送给")[-1].split("，")
                recipient = email_parts[0] if email_parts else "未知"
            else:
                recipient = "test@gmail.com"
            print(f"\n邮件已成功发送到: {recipient}")

    if not email_sent:
        print("\n警告: 未检测到邮件发送结果")

    return final_answer


# 运行主函数
if __name__ == "__main__":
    import asyncio

    asyncio.run(main())