Application de patchs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42from openai import OpenAI

client = OpenAI()

# For brevity, we are including file context in the example input.
# Most agentic use cases should instead equip the model with tools
# for exploring file system state.
RESPONSE_INPUT = """
The user has the following files:
<BEGIN_FILES>
===== lib/fib.py
def fib(n):
    if n <= 1:
        return n
    return fib(n-1) + fib(n-2)

===== run.py
from lib.fib import fib

def main():
  print(fib(42))
<END_FILES>

You are a helpful coding assistant that should assist the user with whatever they
ask.

User query:
Help me rename the fib() function to fibonacci()
"""

response = client.responses.create(
    model="gpt-6-astra",
    input=RESPONSE_INPUT,
    tools=[{"type": "apply_patch"}],
)

# response.output may contain multiple apply_patch_call entries, e.g.:
# - update lib/fib.py
# - update run.py
patch_calls = [
    item.model_dump() for item in response.output if item.type == "apply_patch_call"
]

1
2
3
4
5
6
7
8response, err := client.Responses.New(ctx, responses.ResponseNewParams{
	Model: "gpt-6-astra",
	Input: responses.ResponseNewParamsInputUnion{OfString: openai.String(responseInput)},
	Tools: []responses.ToolUnionParam{{OfApplyPatch: &responses.ApplyPatchToolParam{}}},
})
if err != nil {
	return err
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16import com.openai.client.OpenAIClient;
import com.openai.client.okhttp.OpenAIOkHttpClient;
import com.openai.models.responses.ApplyPatchTool;
import com.openai.models.responses.ResponseCreateParams;

ResponseCreateParams params =
    ResponseCreateParams.builder()
        .model("gpt-6-astra")
        .input(
            "Rename fib() to fibonacci() in lib/fib.py and update run.py to use the new name.")
        .addTool(ApplyPatchTool.builder().build())
        .build();

client.responses().create(params).output().stream()
    .flatMap(item -> item.applyPatchCall().stream())
    .forEach(System.out::println);

1
2
3
4
5
6
7
8
9
10
11
12
13require "openai"

client = OpenAI::Client.new
files = {
  "lib/fib.py" => "def fib(n):\n    if n <= 1:\n        return n\n    return fib(n-1) + fib(n-2)\n",
  "run.py" => "from lib.fib import fib\n\ndef main():\n  print(fib(42))\n"
}
context = files.map { |path, content| "===== #{path}\n#{content}" }.join
response = client.responses.create(
  model: "gpt-6-astra",
  input: "The user has the following files:\n<BEGIN_FILES>\n#{context}<END_FILES>\nHelp me rename fib() to fibonacci() in both existing files.",
  tools: [{ type: :apply_patch }]
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33from apply_patch_harness import apply_operation  # your implementation

for turn in range(9):
    if response.status != "completed":
        raise RuntimeError(f"Patch response ended with status {response.status}")
    if not patch_calls:
        break
    if turn == 8:
        raise RuntimeError("Patch workflow exceeded eight tool turns")

    results = []
    for call in patch_calls:
        success, log_output = apply_operation(call["operation"])
        results.append(
            {
                "type": "apply_patch_call_output",
                "call_id": call["call_id"],
                "status": "completed" if success else "failed",
                "output": log_output,
            }
        )

    response = client.responses.create(
        model="gpt-6-astra",
        previous_response_id=response.id,
        input=results,
        tools=[{"type": "apply_patch"}],
    )
    patch_calls = [
        item.model_dump() for item in response.output if item.type == "apply_patch_call"
    ]

print(response.output_text)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38for turn := 0; ; turn++ {
	if response.Status != responses.ResponseStatusCompleted {
		return fmt.Errorf("response ended with status %s", response.Status)
	}
	patchCalls := make([]responses.ResponseOutputItemUnion, 0)
	for _, item := range response.Output {
		if item.Type == "apply_patch_call" {
			patchCalls = append(patchCalls, item)
		}
	}
	if len(patchCalls) == 0 {
		fmt.Println(response.OutputText())
		return nil
	}
	if turn == 8 {
		return fmt.Errorf("patch workflow exceeded eight tool turns")
	}
	results := make(responses.ResponseInputParam, 0, len(patchCalls))
	for _, call := range patchCalls {
		success, logOutput := applyOperation(files, call.Operation)
		status := "completed"
		if !success {
			status = "failed"
		}
		result := responses.ResponseInputItemParamOfApplyPatchCallOutput(call.CallID, status)
		result.OfApplyPatchCallOutput.Output = openai.String(logOutput)
		results = append(results, result)
	}
	response, err = client.Responses.New(ctx, responses.ResponseNewParams{
		Model:              "gpt-6-astra",
		PreviousResponseID: openai.String(response.ID),
		Input:              responses.ResponseNewParamsInputUnion{OfInputItemList: results},
		Tools:              []responses.ToolUnionParam{{OfApplyPatch: &responses.ApplyPatchToolParam{}}},
	})
	if err != nil {
		return err
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27import com.openai.client.OpenAIClient;
import com.openai.client.okhttp.OpenAIOkHttpClient;
import com.openai.models.responses.ApplyPatchTool;
import com.openai.models.responses.ResponseCreateParams;
import com.openai.models.responses.ResponseInputItem;
import java.util.List;

ResponseCreateParams params =
    ResponseCreateParams.builder()
        .model("gpt-6-astra")
        .inputOfResponse(
            List.of(
                ResponseInputItem.ofApplyPatchCallOutput(
                    ResponseInputItem.ApplyPatchCallOutput.builder()
                        .callId(System.getenv("OPENAI_EXAMPLE_APPLY_PATCH_CALL_ID"))
                        .status(ResponseInputItem.ApplyPatchCallOutput.Status.COMPLETED)
                        .output("Patch applied successfully.")
                        .build())))
        .previousResponseId(System.getenv("OPENAI_EXAMPLE_PREVIOUS_RESPONSE_ID"))
        .addTool(ApplyPatchTool.builder().build())
        .build();

client.responses().create(params).output().stream()
    .flatMap(item -> item.message().stream())
    .flatMap(message -> message.content().stream())
    .flatMap(content -> content.outputText().stream())
    .forEach(text -> System.out.println(text.text()));

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
289.times do |turn|
  raise "Response ended with status #{response.status}" unless response.status == OpenAI::Responses::ResponseStatus::COMPLETED

  patch_calls = response.output.grep(OpenAI::Models::Responses::ResponseApplyPatchToolCall)
  if patch_calls.empty?
    puts(response.output_text)
    break
  end
  raise "Patch workflow exceeded eight tool turns" if turn == 8

  # apply_operation is application-owned; see "Implementing the patch harness" in the guide.
  results = patch_calls.map do |call|
    success, output = apply_operation(files, call.operation)
    {
      type: :apply_patch_call_output,
      call_id: call.call_id,
      status: success ? :completed : :failed,
      output: output
    }
  end
  response = client.responses.create(
    model: "gpt-6-astra",
    previous_response_id: response.id,
    input: results,
    tools: [{ type: :apply_patch }]
  )
end
files.each { |path, content| print("===== #{path}\n#{content}") }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67from pathlib import Path

from agents import Agent, ApplyPatchTool, Runner, apply_diff


class WorkspaceEditor:
    def __init__(self, root="workspace"):
        self.root = Path(root).resolve()
        self.root.mkdir(parents=True, exist_ok=True)

    def _path(self, name):
        path = self.root / name
        resolved = path.resolve()
        if resolved == self.root or not resolved.is_relative_to(self.root):
            raise ValueError("Patch paths must stay inside the workspace")
        return path

    async def create_file(self, operation):
        content = apply_diff("", operation.diff, mode="create")
        path = self._path(operation.path)
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("x", encoding="utf-8") as file:
            file.write(content)
        return {"status": "completed", "output": f"Created {operation.path}"}

    async def update_file(self, operation):
        path = self._path(operation.path)
        content = apply_diff(path.read_text(encoding="utf-8"), operation.diff)
        path.write_text(content, encoding="utf-8")
        return {"status": "completed", "output": f"Updated {operation.path}"}

    async def delete_file(self, operation):
        self._path(operation.path).unlink()
        return {"status": "completed", "output": f"Deleted {operation.path}"}


editor = WorkspaceEditor()

agent = Agent(
    name="Patch Assistant",
    model="gpt-6-astra",
    instructions="Use apply_patch to edit files in the workspace. Supply paths relative to that workspace.",
    tools=[
        ApplyPatchTool(
            editor=editor,
            # could also be a function for you to determine if approval is needed
            needs_approval=True,
            # Implement your own approval logic
            on_approval=lambda _ctx, _approval_item: {"approve": True},
        ),
    ],
)


async def main():
    result = await Runner.run(
        agent,
        input="Create tasks.md with a shopping checklist of 5 entries.",
    )

    print(f"\nFinal response:\n{result.final_output}")


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())