Software 2.0

DSPy Agent开发简明教程

LIBRARY Jan 8, 2025

这篇文章将首先介绍如何使用 DSPy 从头开始​​创建代理。

本文的目的是教育，不要将其用于生产。这是对如何处理它的探索。

我所说的代理是什么意思？行业中有很多不同的定义，我使用...

  • 可以做事的独立（相对）代理
  • 他们可以使用工具。
  • 他们可以计划和执行任务。

在这篇博文中，我将尝试在 DSPy 中构建代理。这实际上更像是一个代理“框架”，但你明白我的意思。

本教程的目标是指导你完成使用 DSPy 构建简单代理应用程序的过程。由于 DSPy 不是代理框架，因此这实际上更像是一篇“DSPy 为你提供什么和不为你提供什么”类型的帖子。

在本篇文章结束时，你将了解计划、工作者和工具等关键概念，以及它们如何协同工作以创建功能性代理系统。

注意：这与 reAct 之类的东西完全不同。这是我们在较低级别从头开始定义代理。

1、设置环境

我们将首先为我们的 DSPy 代理应用程序设置必要的环境。首先，我们将加载自动重新加载扩展以确保我们的代码更改会自动重新加载。

我们将加载环境变量，定义我们的模型。

from dotenv import load_dotenv
load_dotenv()

我正在使用 LLMClient，因此我相应地设置了我的 api_base。

import dspy
import instructor
wrkr = dspy.OpenAI(model='gpt-3.5-turbo', max_tokens=1000, api_base='http://0.0.0.0:4000', model_type="chat")
bss = dspy.OpenAI(model='gpt-4-turbo', max_tokens=1000, api_base='http://0.0.0.0:4000', model_type="chat")
dspy.configure(lm=wrkr) # our default
from typing import List, Any, Callable, Optional
from pydantic import BaseModel

2、定义计划签名

在本节中，我们将介绍“计划”签名的概念。

class Plan(dspy.Signature):
    """Produce a step by step plan to perform the task. 
The plan needs to be in markdown format and should be broken down into big steps (with ## headings) and sub-steps beneath those.
When thinking about your plan, be sure to think about the tools at your disposal and include them in your plan.
    """
    task = dspy.InputField(prefix="Task", desc="The task")
    context = dspy.InputField(format=str, desc="The context around the plan")
    proposed_plan = dspy.OutputField(desc="The proposed, step by step execution plan.")

Plan 类继承自 dspy.Signature，它是定义 DSPy 中的输入和输出字段的基类。该类有三个字段：

  • task：这是一个输入字段，表示需要计划的任务。
  • context：这是另一个输入字段，表示计划周围的上下文。它是 str 类型，因此可以是多行。
  • proposed_plan：这是一个输出字段，将包含分步执行计划。

Plan 类的文档字符串说明，提议的计划应采用 Markdown 格式，顶级步骤用 ## 标题表示，子步骤位于其下方。

此 Plan 类将作为定义代理应用程序中的规划逻辑的基础。

3、创建基本 Worker

我们现在将在我们的代理应用程序中引入 Worker 的概念。Worker 类负责处理分配给它的任务的规划和执行。

我们为 Worker 提供一个角色和 Worker 可以访问的工具列表。我们存储工具描述的字符串表示，这将有助于在规划和执行任务时为 Worker 提供上下文。

我在这里添加了历史记录，尽管我最终没有使用它。我想这是我存储它的地方，但我不是 100% 确定。

forward 方法是 Worker 的主要入口点。

我越想越觉得 forward 其实只是“DSPy 就像 PyTorch”的残余——这实际上可以是任何函数。

我们首先创建一个上下文字符串，其中包含工作者的角色和可用工具的描述。然后，我们将此上下文和任务传递给 plan 方法，该方法返回包含建议计划的结果对象。最后，我们将建议的计划打印到控制台，这样我们就可以看到工作者的“思考过程”。

class Worker(dspy.Module):
    def __init__(self, role:str, tools:List):
        self.role = role
        self.tools = tools
        self.tool_descriptions = "\n".join([f"- {t.name}: {t.description}. To use this tool please provide: `{t.requires}`" for t in tools])
        self.plan = dspy.ChainOfThought(Plan)
    def forward(self, task:str):
        context = f"{self.role}\n{self.tool_descriptions}"
        input_args = dict(
            context = context,
            task = task
        ) # just did args for printing for debugging
        result = self.plan(**input_args)
        print(result.proposed_plan)

4、定义工具

我们现在将在我们的代理应用程序中引入工具的概念。工具是我们的代理可以用来完成任务的专门功能。每个工具都有一个名称、一个描述、一个所需输入列表和一个执行实际工作的函数。Tool 类定义如下：

class Tool(BaseModel):
    name: str
    description: str
    requires: str
    func: Callable

注意：DSPy 不使用函数调用（例如来自 OpenAI），所以我们基本上必须定义自己的函数调用。我不确定最好的方法是什么，但本着“黑客马拉松”的精神，我就这样做了。

大多数参数都是不言自明的，但 requires字段指定了工具运行所需的输入。而 func字段是一个可调用函数，它接受所需的输入并返回使用该工具的结果。

为了演示工具的工作原理，让我们创建一些测试工具：

test_tools = [
    Tool(name="phone", description="a way of making phone calls", requires="phone_number", func=lambda x: "they've got time"),
    Tool(name="local business lookup", description="Look up businesses by category", requires="business category", func=lambda x: "Bills landscaping: 415-555-5555")
]

然后我们可以在代理应用程序中使用这些工具。例如，我们可以创建一个 Worker 并让它使用测试工具来尝试完成一项任务：

with dspy.context(lm=wrkr):
    Worker("assistant", test_tools).forward("get this yard cleaned up.")
# you'll see output you might expect.

这演示了如何将 Tools 集成到代理应用程序中以提供可用于完成任务的专门功能。在下一节中，我们将尝试将其提升到一个新的水平。

5、改进 Worker 类

我们引入了基本的 Worker 类，但为了使我们的代理应用程序更加健壮，我们需要改进 Worker 类。在更新的 Worker2 类中，我们添加了两个新方法：plan 和 execute。

plan 方法负责为给定任务生成拟议的行动计划。

execute 方法负责执行计划的操作。它首先检查是否可以在不使用任何工具的情况下完成任务。如果是，它只会返回一条成功消息。否则，它再次使用 DSPy 中的 ChainOfThought 模块来确定需要哪个工具以及为该工具提供哪些参数。然后它调用适当工具的 func 方法来完成任务。

注意：我不太确定这是否是正确的选择。它可以工作，但在我真正称其为“好”之前，它需要更多的思考。

通过使用 ChainOfThought 模块进行规划和执行，我们让工作人员能够仔细思考问题，将其分解为更小的步骤，然后选择正确的工具来完成每个步骤。这里最棒的是，我们可以内联地或以正式签名的形式定义它们。

class Worker2(dspy.Module):
    def __init__(self, role:str, tools:List):
        self.role = role
        self.tools = dict([(t.name, t) for t in tools])
        self.tool_descriptions = "\n".join([f"- {t.name}: {t.description}. To use this tool please provide: `{t.requires}`" for t in tools])
        self._plan = dspy.ChainOfThought(Plan)
        self._tool = dspy.ChainOfThought("task, context -> tool_name, tool_argument")
        
        print(self.tool_descriptions)
    def plan(self, task:str, feedback:Optional[str]=None):
        context = f"Your role:{self.role}\n Tools at your disposal:\n{self.tool_descriptions}"
        if feedback:
            context += f"\nPrevious feedback on your prior plan {feedback}"
        input_args = dict(
            task=task,
            context=context
        )    
        result = self._plan(**input_args)
        return result.proposed_plan
    def execute(self, task:str, use_tool:bool):
        print(f"executing {task}")
        if not use_tool:
            return f"{task} completed successfully"
            
        res = self._tool(task=task, context=self.tool_descriptions)
        t = res.tool_name
        arg = res.tool_argument
        if t in self.tools:
            complete = self.tools[t].func(arg)
            return complete
        return "Not done"

总体而言，对 Worker 类的改进使代理应用程序更加智能和灵活，为更高级的功能奠定了基础

后续章节中将介绍如何创建专用工具。

6、创建专用工具

对于助理，我们创建了两个工具：email_tool 和 schedule_meeting_tool。email_tool 允许助理发送和接收电子邮件，而 schedule_meeting_tool 可用于安排会议。这两个工具都有一个名称、一个描述、一个指定所需输入的需要字段和一个定义工具功能的函数字段。

email_tool = Tool(
    name="email",
    description="Send and receive emails",
    requires="email_address",
    func=lambda x: f"Email sent to {x}"
)
schedule_meeting_tool = Tool(
    name="schedule meeting",
    description="Schedule meetings",
    requires="meeting_details",
    func=lambda x: f"Meeting scheduled on {x}"
)

请注意，我们为看门人添加了一个边缘案例，他们需要报告维护问题。

cleaning_supplies_tool = Tool(
    name="cleaning supplies",
    description="List of cleaning supplies needed",
    requires="cleaning_area",
    func=lambda x: f"Need supplies for {x}"
)
maintenance_report_tool = Tool(
    name="maintenance report",
    description="Report maintenance issues",
    requires="issue_description",
    func=lambda x: f"There's too much work for one person. I need help!"
)

对于软件工程师，我们提供了一个 code_compiler_tool 来编译代码，以及一个 bug_tracker_tool 来跟踪和报告错误。

code_compiler_tool = Tool(
    name="code compiler",
    description="Compile code",
    requires="source_code",
    func=lambda x: "Code compiled successfully"
)
bug_tracker_tool = Tool(
    name="bug tracker",
    description="Track and report bugs",
    requires="bug_details",
    func=lambda x: f"Bug reported: {x}"
)

最后，我们为厨师创建了工具，包括用于查找食谱的 recipe_lookup_tool 和用于检查厨房库存的 kitchen_inventory_tool。

recipe_lookup_tool = Tool(
    name="recipe lookup",
    description="Look up recipes",
    requires="dish_name",
    func=lambda x: f"Recipe for {x} found"
)
kitchen_inventory_tool = Tool(
    name="kitchen inventory",
    description="Check kitchen inventory",
    requires="ingredient",
    func=lambda x: f"Inventory checked for {x}"
)

定义这些专用工具后，我们将它们分配给相应的工人。这样，每个工人都可以访问执行特定任务所需的工具。

workers = [
    Worker2("assistant", [email_tool, schedule_meeting_tool]),
    Worker2("janitor", [cleaning_supplies_tool, maintenance_report_tool]),
    Worker2("software engineer", [code_compiler_tool, bug_tracker_tool]),
    Worker2("cook", [recipe_lookup_tool, kitchen_inventory_tool])
]

通过提供这些示例，我们演示了如何创建各种工具来支持代理应用程序中的不同角色和功能。

7、解析计划

有人会因为我写这个而诧异，但是是的……我使用了 Instructor。Instructor 非常适合这种事情。我可以用正则表达式或字符串格式解析计划，但 Instructor 允许我做同样的事情，而不必处理细节/细微差别。

为了实际解析计划，我们将使用 Instructor 库，它提供了一个方便的界面，用于与 GPT-3.5 等大型语言模型进行交互。get_plan 函数接收计划文本和一些其他上下文，然后使用 Instructor 生成我们可以使用的 ParsedPlan 对象。这使我们能够轻松地从计划中提取单个任务和子任务，为代理应用程序中的后续步骤奠定基础。

from pydantic import Field
import instructor
from openai import OpenAI
_client = instructor.from_openai(OpenAI(base_url="http://0.0.0.0:4000/"))
class SubTask(BaseModel):
    action:str
    assignee: str
    requires_tool: bool = Field(..., description="Does this require the use of a specific tool?")
                               
class Task(BaseModel):
    sub_tasks:List[SubTask]
    
class ParsedPlan(BaseModel):
    tasks: List[Task]
def get_plan(plan:str, context:str):
    return _client.chat.completions.create(
        response_model=ParsedPlan,
        model="gpt-3.5-turbo",
        messages=[
            dict(role="system", content="You help parse markdown into a structured format."),
            dict(role="user", content=f"Here is the context about the plan including the available tools: \n{context} \n\n The plan: \n\n {plan}")
        ],
    )

8、实现 Boss 类

Boss 是我们代理应用程序中的一个关键组件，负责监督代理（或工作者）的工作并确保正确分配和执行任务。让我们深入了解 Boss 类的实现。

我们使用基本上下文、直接报告（工作者）列表和语言模型初始化“Boss”对象。基本上下文为 Boss 的决策提供了整体上下文，直接报告存储在字典中以方便访问。

我们还生成了一份关于 Boss 能力的报告，每个直接报告并将其存储在 report_capabilities 属性中 - 这基本上只是让“老板”了解员工可以做什么。

class Boss(dspy.Module):
    def __init__(self, base_context:str, direct_reports=List, lm=bss):
        self.base_context = base_context
        self._plan = dspy.ChainOfThought("task, context -> assignee")
        self._approve = dspy.ChainOfThought("task, context -> approve")
        self._critique = dspy.ChainOfThought("task, context -> critique")
        self.reports = dict((d.role,d) for d in direct_reports)
        self.lm = lm
        report_capabilities = []
        for r in direct_reports:
            report_capabilities.append(f"{r.role} has the follow tools:\n{r.tool_descriptions}")
        self.report_capabilities = "\n".join(report_capabilities) 
        print(self.report_capabilities)
    # The `critique` method allows the Boss to provide feedback on a proposed plan. It takes the task, the proposed plan, and an optional extra context as input, and uses the `_critique` chain of thought to generate a critique.
    def critique(self, task:str, plan:str, extra_context:Optional[str]=None):
        context=self.base_context
        if extra_context:
            context += "\n"
            context += extra_context
        
        crit_args = dict(
            context=context,
            task=task,
            proposed_plan=plan)
        with dspy.context(lm=self.lm):
            result = self._critique(**crit_args)
        return result.critique
    # The `approve` method allows the Boss to approve a proposed plan. It takes the task, the proposed plan, and an optional extra context as input, and uses the `_approve` chain of thought to generate an approval decision.
    def approve(self, task:str, plan:str, extra_context:Optional[str]=None):
        context=self.base_context + "\n You only approve plans after 2 iterations"
        if extra_context:
            context += "\n"
            context += extra_context
        
        approval_args = dict(
            context=context,
            task=task,
            proposed_plan=plan)
        with dspy.context(lm=self.lm):
            result = self._approve(**approval_args)
        return result.approve        
    # The `plan` method is the core of the Boss class. It takes a task as input and uses the `_plan` chain of thought to determine which of the direct reports should be assigned to the task. The method then iterates through the assignment process, providing feedback and critiques until a suitable assignee is found. Once the assignee is determined, the Boss calls the `plan` method of the assigned worker to generate a plan for the task. The Boss then approves the plan, providing critiques and feedback as necessary, until the plan is approved.
    def plan(self, task:str):
        # note: this function is a mess, don't judge me
        # I haven't built an agent framework before, so I'm not sure of the ergonomics
        # and best approach
        context=self.base_context + f"Here are your direct report capabilities: {self.report_capabilities}"
        
        plan_args = dict(
            context = context,
            task=f"Which person should take on the following task: {task}"
        )
        assignee = self._plan(**plan_args).assignee
        is_assigned = assignee.lower() in self.reports
        report = None
        print("assigning")
        for x in range(3):
            if is_assigned:
                report = self.reports[assignee]
            else:
                context += f"\n\n you tried to assign to {assignee} but that's not a valid one. Think carefully and assign the proper report"
                plan_args = dict(
                    context = context,
                    task=f"Which person should take on the following task: {task}"
                )
                assignee = self._plan(**plan_args).assignee
        assert report, "Failed to assign"
        print("assigning complete")
        print("planning")
        reports_plan = report.plan(task)
        with dspy.context(lm=self.lm):
            approval = self.approve(task, reports_plan)
            is_approved = "yes" in approval.lower() and "no" not in approval.lower()
        
        for x in range(2): # I created cycles to simulate failures, this might be a while loop in production
            print(f"Cycle {x}: {approval}")
            if is_approved:
                break
            feedback = self.critique(task, reports_plan)
            feedback = f"Prior plan: {reports_plan}\n Boss's Feedback: {feedback}"
            print(feedback)
            reports_plan = report.plan(task, feedback)
            print("new plan===>")
            print(reports_plan)
            complete = f"{feedback}\n\nNew plan:\n\n{reports_plan}"
            approval = self.approve(task, reports_plan)
            is_approved = "yes" in approval.lower()
        print("Now working")
        parsed_plan = get_plan(reports_plan, f"The assignee is: {assignee}. The rest of the team is: {self.report_capabilities}")
        results = []
        for task in parsed_plan.tasks:
            for sub_task in task.sub_tasks:    
                task_result = self.reports[sub_task.assignee].execute(sub_task.action, sub_task.requires_tool)
                results.append(f"\n{task_result}: {sub_task.action}\n")
        print("end result")
        print("\n".join(results))

Boss 类的这个实现演示了 Boss 和工人之间的交互，以及 Boss 如何提供监督和指导以确保正确规划和执行任务。

这里有更多复杂的机会，但我们只是在构建一个代理框架。

9、将它们整合在一起

现在我们已经介绍了代理应用程序的关键组件，让我们将它们放在一起并在动作。我们将创建一个 Boss 类的实例，并演示如何使用 plan 方法将任务分配给工人并执行计划。

首先，我们将创建一个 Boss 实例并传入我们之前定义的工人：

b = Boss("You are a boss that manages a team of people, you're responsible for them doing well and completing the tasks you are given.", workers)

在这个例子中，我们传入了老板的角色和职责的描述，以及老板将管理的工人列表。

461：接下来，我们将调用老板实例中的 plan 方法并提供要完成的任务：

b.plan("clean up the yard")

当我们调用 plan 方法时，老板将解析任务，将子任务分配给适当的工人，并协调计划的执行。

老板为工人提供有关计划的反馈。

这里很酷的是，老板比工人拥有更多的背景信息（例如，更强大的模型）。

此示例演示了我们的代理应用程序的核心功能，但仍有很大的优化和扩展潜力。我们将在结论中探讨其中的一些可能性。

10、结束语

好吧，这花了几个小时才完成。我不确定这是否是最好的方法，但这是一种方法。

这一切都很棒，这里没有什么疯狂的。我没有展示但我希望有时间展示的东西是通过优化“训练”这些代理的能力。你可以想象运行一个 DSPy 优化，使用 bootstrap 优化每个模型。

这需要做更多的工作，但这就是 DSPy 的真正力量所在。

另一种方法

LLMClient 是 Typescript 中基于 DSPy 的框架。它以不同的方式解决问题，但它是一个很好的例子，说明你可以如何做到这一点。

通过基本上为每个代理提供单个签名，你可以创建一个稍微简单的系统。我在编写自己的内容后发现了这一点，但如果读者感兴趣的话，一定要查看一下。

const researcher = new Agent(ai, {
  name: 'researcher',
  description: 'Researcher agent',
  signature: `physicsQuestion "physics questions" -> answer "reply in bullet points"`
});
const summarizer = new Agent(ai, {
  name: 'summarizer',
  description: 'Summarizer agent',
  signature: `text "text so summarize" -> shortSummary "summarize in 5 to 10 words"`
});
const agent = new Agent(ai, {
  name: 'agent',
506: description: 'An agent to research complex topics'
  signature: `question -> answer`,
  agents: [researcher, summarizer]
});
agent.forward({ questions: "How many atoms are there in the universe" })
实际结论

在本教程中，我们通过构建一个基本的代理应用程序迈出了进入 DSPy 代理世界的第一步。我借鉴了一些现有的代理框架（如 CrewAI）。

DSPy 的低抽象开销为我们在设计和定制代理应用程序方面提供了很大的灵活性，但这也意味着更多的决策。

像 CrewAI 这样的东西会“感觉”简单得多。

虽然我们已经涵盖了很多内容，但仍有大量的探索和改进空间

本教程的妙处在于，您可以通过阅读和使用代码来学到很多东西。

不那么容易的部分是，要创建一个真正强大且功能强大的代理应用程序，还有很多工作要做。但这也是乐趣的一部分，对吧？

原文链接：A first attempt at DSPy Agents from scratch

汇智网翻译整理，转载请标明出处

Tags

admin

Recommended for you