分解复杂任务
分解式 Prompting1 利用大型语言模型 (LLM) 将复杂任务分解为一系列可管理的子任务。每个子任务随后由特定函数处理，使 LLM 能够更有效、更有系统地处理复杂问题。
在下面的代码片段中，我们定义了一系列数据模型和函数来实现这种方法。
derive_action_plan 函数使用 LLM 生成行动计划，然后逐步执行该计划。每个行动可以是
- InitialInput: 表示我们需要处理的原始 Prompt 的一部分
- Split : 使用给定分隔符分割字符串的操作
- StrPos: 给定索引提取字符串的操作
- Merge: 使用给定字符将字符串列表合并的操作
我们可以使用 instructor 实现这一点，如下所示。
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Union
client = instructor.from_openai(OpenAI())
class Split(BaseModel):
split_char: str = Field(
description="""This is the character to split
the string with"""
)
def split_chars(self, s: str, c: str):
return s.split(c)
class StrPos(BaseModel):
index: int = Field(
description="""This is the index of the character
we wish to return"""
)
def get_char(self, s: list[str], i: int):
return [c[i] for c in s]
class Merge(BaseModel):
merge_char: str = Field(
description="""This is the character to merge the
inputs we plan to pass to this function with"""
)
def merge_string(self, s: list[str]):
return self.merge_char.join(s)
class Action(BaseModel):
id: int = Field(
description="""Unique Incremental id to identify
this action with"""
)
action: Union[Split, StrPos, Merge]
class ActionPlan(BaseModel):
initial_data: str
plan: list[Action]
def derive_action_plan(task_description: str) -> ActionPlan:
return client.chat.completions.create(
messages=[
{
"role": "system",
"content": """Generate an action plan to help you complete
the task outlined by the user""",
},
{"role": "user", "content": task_description},
],
response_model=ActionPlan,
max_retries=3,
model="gpt-4o",
)
if __name__ == "__main__":
task = """Concatenate the second letter of every word in Jack
Ryan together"""
plan = derive_action_plan(task)
print(plan.model_dump_json(indent=2))
"""
{
"initial_data": "Jack Ryan",
"plan": [
{
"id": 1,
"action": {
"split_char": " "
}
},
{
"id": 2,
"action": {
"index": 1
}
},
{
"id": 3,
"action": {
"merge_char": ""
}
}
]
}
"""
curr = plan.initial_data
cache = {}
for action in plan.plan:
if isinstance(action.action, Split) and isinstance(curr, str):
curr = action.action.split_chars(curr, action.action.split_char)
elif isinstance(action.action, StrPos) and isinstance(curr, list):
curr = action.action.get_char(curr, action.action.index)
elif isinstance(action.action, Merge) and isinstance(curr, list):
curr = action.action.merge_string(curr)
else:
raise ValueError("Unsupported Operation")
print(action, curr)
#> id=1 action=Split(split_char=' ') ['Jack', 'Ryan']
#> id=2 action=StrPos(index=1) ['a', 'y']
#> id=3 action=Merge(merge_char='') ay
print(curr)
#> ay