入门学习

该模式主要思想是让llm按照规定进行任务拆解，然后按照任务进行输出，并把每次输出穿起来，按顺序输出给下一个任务，最后任务结束

工作准备：llmclient配置函数，下方的main是测试用的：

1、配置api（略）

import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

load_dotenv()


class HelloAgentsLLM:

    def __init__(self, model: str = None, apiKey: str = None, baseURL: str = None, timeout: int = None):
        self.model = model or os.getenv("MODEL_ID")
        self.apiKey = apiKey or os.getenv("ARK_API_KEY")
        self.baseURL = baseURL or os.getenv("BASE_URL")
        self.timeout = timeout or int(os.getenv("TIMEOUT", 60))

        if not all([self.model, self.apiKey, self.baseURL]):
            raise ValueError(
                "Model, API key, and Base URL must be provided either as arguments or in the .env file.")

        # 初始化一个客户端对象
        self.client = OpenAI(api_key=self.apiKey,
                             base_url=self.baseURL, timeout=self.timeout)

    def think(self, message: List[Dict[str, str]], temperature: float = 0.5) -> str:

        print(f"正在调用{self.model}模型。。。")

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=message,
                temperature=temperature,
                stream=True
            )
            print("模型调用完成！")
            collected_content = []
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True)
                collected_content.append(content)
            print()  # 换行
            return "".join(collected_content)
        except Exception as e:
            print(f"调用模型时发生错误: {e}")
            return None


if __name__ == "__main__":
    try:
        llmClient = HelloAgentsLLM()

        exampleMessages = [
            {"role": "system",
                "content": "You are a helpful assistant that writes Python code."},
            {"role": "user", "content": "写一个快速排序算法"}
        ]

        print("_____调用LLM模型_____")
        responseText = llmClient.think(exampleMessages, temperature=0.5)
        if (responseText):
            print("模型返回的内容：")
            print(responseText)
    except Exception as e:
        print(f"发生错误: {e}")

2、配置prompt，相对于react模式，这个模式的promt分为任务拆解规则，和执行器规则，规定好planner和executor

PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]
```
"""

EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。

# 原始问题:
{question}

# 完整计划:
{plan}

# 历史步骤与结果:
{history}

# 当前步骤:
{current_step}

请仅输出针对“当前步骤”的回答:
"""

2、设置planner和executor：

planner对大模型输出的plan进行字符串转换并接收，调用llmclient中的thingk函数即可：

import sys
import re
import os
from pathlib import Path
import ast
rootPath = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(rootPath))

from client.LLMClient import HelloAgentsLLM
from prompt.plannerPrompt import PLANNER_PROMPT_TEMPLATE

class Planner:
    def __init__(self, llm_client):
        self.llm_client = llm_client

    def plan(self, question: str) -> list[str]:
        """根据用户问题生成一个行动计划"""
        prompt = PLANNER_PROMPT_TEMPLATE.format(question = question)
        messages = [{"role": "user","content":prompt}]

        print("_____调用LLM模型生成计划_____")
        response_text = self.llm_client.think(message=messages) or ""
        print(f"计划已生成:\n{response_text}")


        #解析LLM输出的列表字符串
        try:
            # 找到 '''python'''之间的内容
            plan_str = response_text.split("```python")[1].split("```")[0].strip()
            plan = ast.literal_eval(plan_str)
            return plan if isinstance(plan, list) else []
        except (ValueError, SyntaxError, IndexError) as e:
            print(f"解析计划时出错：{e}")
            print(f"原始响应：{response_text}")
            return []
        except Exception as e:
            print(f"解析时出现未知错误：{e}")
            return []

编写Executor：
 

import sys
from pathlib import Path

rootPath = Path(__file__).resolve().parent.parent
sys.path.insert(0,str(rootPath))

from prompt.ExecutorPrompt import EXECUTOR_PROMPT_TEMPLATE

class Executor:

    def __init__(self, llm_client):
        self.llm_client = llm_client

    def execute(self, question: str, plan: list[str]) -> str:
        history = ""

        print("\n ---正在执行计划---")
        for i, step in enumerate(plan):
            print(f"\n 正在执行步骤{i+1}")
            prompt = EXECUTOR_PROMPT_TEMPLATE.format(
                question = question,
                plan = plan,
                history = history if history else "无",
                current_step = step
            )
            messages = [{"role":"user","content": prompt}]
            response_text = self.llm_client.think(message = messages) or ""
            history += f"步骤{i+1}：{step}\n结果：{response_text}\n\n"
            print(f"步骤{i+1}已完成,结果：{response_text}")
        
        final_answer = response_text
        return final_answer

3、编写agent，把llm模型调用，planner，executor集合在一起

from pathlib import Path
import sys
rootPath = Path(__file__).absolute().parent.parent
sys.path.insert(0,str(rootPath))

from planner.planAndSolvePlanner import Planner
from Executor.executor import Executor

class PlanAndSolveAgent:

    def __init__(self,llm_client):
        self.llm_client = llm_client
        self.planner = Planner(self.llm_client)
        self.executor = Executor(self.llm_client)

    def run(self, question: str):
        """
        运行智能体完整流程，包括plan和executor
        """
        print(f"\n---开始处理问题---\n问题:{question}")

        #1、调用planner生成计划
        plan = self.planner.plan(question)

        if not plan:
            print("\n---任务终止---\n无法生成有效的行动计划")
            return
        
        #2、调用execute执行器
        final_answer = self.executor.execute(question, plan)
        print(f"\n---任务完成---\n最终答案：{final_answer}")

之后写一个main函数调用即可