文档API 参考📓 教程🧑‍🍳 食谱🤝 集成💜 Discord🎨 Studio
API 参考

GitHub

Haystack GitHub 集成

模块 haystack_integrations.components.connectors.github.file_editor

命令

用于 GitHub 文件操作的可用命令。

属性:

  • EDIT - 编辑现有文件，替换内容
  • UNDO - 撤销上次由同一用户提交的更改
  • CREATE - 创建一个新文件
  • DELETE - 删除现有文件

GitHubFileEditor

一个用于编辑 GitHub 存储库中文件的 Haystack 组件。

支持通过 GitHub API 编辑、撤销更改、删除文件和创建新文件。

使用示例

from haystack_integrations.components.connectors.github import Command, GitHubFileEditor
from haystack.utils import Secret

# Initialize with default repo and branch
editor = GitHubFileEditor(
    github_token=Secret.from_env_var("GITHUB_TOKEN"),
    repo="owner/repo",
    branch="main"
)

# Edit a file using default repo and branch
result = editor.run(
    command=Command.EDIT,
    payload={
        "path": "path/to/file.py",
        "original": "def old_function():",
        "replacement": "def new_function():",
        "message": "Renamed function for clarity"
    }
)

# Edit a file in a different repo/branch
result = editor.run(
    command=Command.EDIT,
    repo="other-owner/other-repo",  # Override default repo
    branch="feature",  # Override default branch
    payload={
        "path": "path/to/file.py",
        "original": "def old_function():",
        "replacement": "def new_function():",
        "message": "Renamed function for clarity"
    }
)

GitHubFileEditor.__init__

def __init__(*,
             github_token: Secret = Secret.from_env_var("GITHUB_TOKEN"),
             repo: Optional[str] = None,
             branch: str = "main",
             raise_on_failure: bool = True)

初始化组件。

参数:

  • github_token: 用于 API 身份验证的 GitHub 个人访问令牌
  • repo: 默认存储库，格式为 owner/repo
  • branch: 要使用的默认分支
  • raise_on_failure: 如果为 True，则在 API 错误时引发异常

引发:

  • TypeError: 如果 github_token 不是 Secret 类型

GitHubFileEditor.run

@component.output_types(result=str)
def run(command: Union[Command, str],
        payload: Dict[str, Any],
        repo: Optional[str] = None,
        branch: Optional[str] = None) -> Dict[str, str]

处理 GitHub 文件操作。

参数:

  • command: 要执行的操作（“edit”、“undo”、“create”、“delete”）
  • payload: 包含命令特定参数的字典
  • repo: 要进行操作的存储库，格式为 owner/repo（如果提供，则覆盖默认值）
  • branch: 要进行操作的分支（如果提供，则覆盖默认值）

引发:

  • ValueError: 如果命令不是有效的 Command 枚举值

返回值:

包含操作结果的字典

GitHubFileEditor.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

GitHubFileEditor.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GitHubFileEditor"

从字典中反序列化组件。

模块 haystack_integrations.components.connectors.github.issue_commenter

GitHubIssueCommenter

将评论发布到 GitHub 问题。

该组件接收一个 GitHub 问题 URL 和评论文本，然后使用 GitHub API 将评论发布到指定的问题。

使用示例

from haystack_integrations.components.connectors.github import GitHubIssueCommenter
from haystack.utils import Secret

commenter = GitHubIssueCommenter(github_token=Secret.from_env_var("GITHUB_TOKEN"))
result = commenter.run(
    url="https://github.com/owner/repo/issues/123",
    comment="Thanks for reporting this issue! We'll look into it."
)

print(result["success"])

GitHubIssueCommenter.__init__

def __init__(*,
             github_token: Secret = Secret.from_env_var("GITHUB_TOKEN"),
             raise_on_failure: bool = True,
             retry_attempts: int = 2)

初始化组件。

参数:

  • github_token: 用于 API 身份验证的 GitHub 个人访问令牌（作为 Secret）
  • raise_on_failure: 如果为 True，则在 API 错误时引发异常
  • retry_attempts: 失败请求的重试次数

GitHubIssueCommenter.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

GitHubIssueCommenter.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GitHubIssueCommenter"

从字典中反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

GitHubIssueCommenter.run

@component.output_types(success=bool)
def run(url: str, comment: str) -> dict

将评论发布到 GitHub 问题。

参数:

  • url: GitHub 问题 URL
  • comment: 要发布的评论文本

返回值:

包含成功状态的字典

模块 haystack_integrations.components.connectors.github.issue_viewer

GitHubIssueViewer

获取并解析 GitHub 问题到 Haystack 文档。

该组件接收一个 GitHub 问题 URL，并返回一个文档列表，其中

  • 第一个文档包含主要问题内容
  • 后续文档包含问题评论

使用示例

from haystack_integrations.components.connectors.github import GitHubIssueViewer

viewer = GitHubIssueViewer()
docs = viewer.run(
    url="https://github.com/owner/repo/issues/123"
)["documents"]

print(docs)

GitHubIssueViewer.__init__

def __init__(*,
             github_token: Optional[Secret] = None,
             raise_on_failure: bool = True,
             retry_attempts: int = 2)

初始化组件。

参数:

  • github_token: 用于 API 身份验证的 GitHub 个人访问令牌（作为 Secret）
  • raise_on_failure: 如果为 True，则在 API 错误时引发异常
  • retry_attempts: 失败请求的重试次数

GitHubIssueViewer.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

GitHubIssueViewer.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GitHubIssueViewer"

从字典中反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

GitHubIssueViewer.run

@component.output_types(documents=List[Document])
def run(url: str) -> dict

处理 GitHub 问题 URL 并返回文档。

参数:

  • url: GitHub 问题 URL

返回值:

包含文档列表的字典

模块 haystack_integrations.components.connectors.github.pr_creator

GitHubPRCreator

一个 Haystack 组件，用于从 fork 创建拉取请求回原始存储库。

使用已认证用户的 fork 创建 PR，并将其链接到现有问题。

使用示例

from haystack_integrations.components.connectors.github import GitHubPRCreator
from haystack.utils import Secret

pr_creator = GitHubPRCreator(
    github_token=Secret.from_env_var("GITHUB_TOKEN")  # Token from the fork owner
)

# Create a PR from your fork
result = pr_creator.run(
    issue_url="https://github.com/owner/repo/issues/123",
    title="Fix issue `123`",
    body="This PR addresses issue `123`",
    branch="feature-branch",     # The branch in your fork with the changes
    base="main"                  # The branch in the original repo to merge into
)

GitHubPRCreator.__init__

def __init__(*,
             github_token: Secret = Secret.from_env_var("GITHUB_TOKEN"),
             raise_on_failure: bool = True)

初始化组件。

参数:

  • github_token: 用于身份验证的 GitHub 个人访问令牌（来自 fork 的所有者）
  • raise_on_failure: 如果为 True，则在 API 错误时引发异常

GitHubPRCreator.run

@component.output_types(result=str)
def run(issue_url: str,
        title: str,
        branch: str,
        base: str,
        body: str = "",
        draft: bool = False) -> Dict[str, str]

从您的 fork 创建一个新的拉取请求到原始存储库，并链接到指定的问题。

参数:

  • issue_url: 要链接 PR 的 GitHub 问题 URL
  • title: 拉取请求的标题
  • branch: 您 fork 中实现更改的分支名称
  • base: 您希望合并到的原始存储库中的分支名称
  • body: 拉取请求描述的附加内容
  • draft: 是否创建草稿拉取请求

返回值:

包含操作结果的字典

GitHubPRCreator.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

GitHubPRCreator.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GitHubPRCreator"

从字典中反序列化组件。

模块 haystack_integrations.components.connectors.github.repo_viewer

GitHubItem

表示 GitHub 存储库中的一项（文件或目录）

type

"file" 或 "dir"

GitHubRepoViewer

导航并从 GitHub 存储库中获取内容。

对于目录

  • 返回一个文档列表，每个文档代表一个项目
  • 每个文档的内容是项目名称
  • 完整路径和元数据在 Document.meta 中

对于文件

  • 返回一个文档
  • 文档的内容是文件内容
  • 完整路径和元数据在 Document.meta 中

对于错误

  • 返回一个文档
  • 文档的内容是错误消息
  • 文档的 meta 包含 type="error"

使用示例

from haystack_integrations.components.connectors.github import GitHubRepoViewer

viewer = GitHubRepoViewer()

# List directory contents - returns multiple documents
result = viewer.run(
    repo="owner/repository",
    path="docs/",
    branch="main"
)
print(result)

# Get specific file - returns single document
result = viewer.run(
    repo="owner/repository",
    path="README.md",
    branch="main"
)
print(result)

GitHubRepoViewer.__init__

def __init__(*,
             github_token: Optional[Secret] = None,
             raise_on_failure: bool = True,
             max_file_size: int = 1_000_000,
             repo: Optional[str] = None,
             branch: str = "main")

初始化组件。

参数:

  • github_token: 用于 API 身份验证的 GitHub 个人访问令牌
  • raise_on_failure: 如果为 True，则在 API 错误时引发异常
  • max_file_size: 要获取的最大文件大小（字节，默认值：1MB）
  • repo: 存储库，格式为 "owner/repo"
  • branch: 要使用的 Git 引用（分支、标签、提交）

GitHubRepoViewer.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

GitHubRepoViewer.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GitHubRepoViewer"

从字典中反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

GitHubRepoViewer.run

@component.output_types(documents=List[Document])
def run(path: str,
        repo: Optional[str] = None,
        branch: Optional[str] = None) -> Dict[str, List[Document]]

处理 GitHub 存储库路径并返回文档。

参数:

  • repo: 存储库，格式为 "owner/repo"
  • path: 存储库中的路径（默认：根目录）
  • branch: 要使用的 Git 引用（分支、标签、提交）

返回值:

包含文档列表的字典

模块 haystack_integrations.components.connectors.github.repo_forker

GitHubRepoForker

从问题 URL fork 一个 GitHub 存储库。

该组件接收一个 GitHub 问题 URL，提取存储库信息，创建或同步该存储库的 fork，并可选地创建一个特定于问题的分支。

使用示例

from haystack_integrations.components.connectors.github import GitHubRepoForker
from haystack.utils import Secret

# Using direct token with auto-sync and branch creation
forker = GitHubRepoForker(
    github_token=Secret.from_env_var("GITHUB_TOKEN"),
    auto_sync=True,
    create_branch=True
)

result = forker.run(url="https://github.com/owner/repo/issues/123")
print(result)
# Will create or sync fork and create branch "fix-123"

GitHubRepoForker.__init__

def __init__(*,
             github_token: Secret = Secret.from_env_var("GITHUB_TOKEN"),
             raise_on_failure: bool = True,
             wait_for_completion: bool = False,
             max_wait_seconds: int = 300,
             poll_interval: int = 2,
             auto_sync: bool = True,
             create_branch: bool = True)

初始化组件。

参数:

  • github_token: 用于 API 身份验证的 GitHub 个人访问令牌
  • raise_on_failure: 如果为 True，则在 API 错误时引发异常
  • wait_for_completion: 如果为 True，则等待 fork 完全创建
  • max_wait_seconds: 等待 fork 完成的最大秒数
  • poll_interval: 状态检查之间的间隔（秒）
  • auto_sync: 如果为 True，并且 fork 已存在，则将其与原始存储库同步
  • create_branch: 如果为 True，则基于问题编号创建一个修复分支

GitHubRepoForker.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

GitHubRepoForker.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GitHubRepoForker"

从字典中反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

GitHubRepoForker.run

@component.output_types(repo=str, issue_branch=str)
def run(url: str) -> dict

处理 GitHub 问题 URL 并创建或同步存储库的 fork。

参数:

  • url: GitHub 问题 URL

返回值:

包含存储库路径（owner/repo 格式）的字典