Dynamiq

Dynamiq 是一个用于 Agentic AI 和 LLM 应用的编排框架。

欢迎来到 Dynamiq！🤖

Dynamiq 是你的一站式生成式 AI (Gen AI) 框架，旨在简化 AI 驱动型应用程序的开发。Dynamiq 专注于编排检索增强生成 (RAG) 和大语言模型 (LLM) Agent。

入门指南

准备好开始了吗？以下是如何开始使用 Dynamiq 的方法：

安装

首先，让我们安装 Dynamiq。你需要安装 Python，请确保你的机器上已正确配置。然后运行：

pip install dynamiq

或者从源代码构建 Python 包：

git clone https://github.com/dynamiq-ai/dynamiq.git
cd dynamiq
poetry install

文档

如需更多示例和详细指南，请参考我们的文档。

示例

简单的 LLM 工作流

这是一个帮助你开始使用 Dynamiq 的简单示例：

from dynamiq.nodes.llms.openai import OpenAI
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.prompts import Prompt, Message

# 定义翻译任务的 Prompt 模板
prompt_template = """
Translate the following text into English: {{ text }}
"""

# 使用定义的模板创建 Prompt 对象
prompt = Prompt(messages=[Message(content=prompt_template, role="user")])

# 设置你的 LLM（大语言模型）节点
llm = OpenAI(
    id="openai",  # 节点的唯一标识符
    connection=OpenAIConnection(api_key="OPENAI_API_KEY"),  # 使用 API Key 连接
    model="gpt-4o",  # 使用的模型
    temperature=0.3,  # 模型的采样温度
    max_tokens=1000,  # 输出的最大 token 数
    prompt=prompt  # 模型使用的 Prompt
)

# 使用输入数据运行 LLM 节点
result = llm.run(
    input_data={
        "text": "Hola Mundo!"  # 待翻译文本
    }
)

# 打印翻译结果
print(result.output)

带有异步执行的简单 ReAct Agent

这是一个拥有 E2B 代码解释器访问权限，并能够解决复杂编码任务的 Agent。

from dynamiq.nodes.llms.openai import OpenAI
from dynamiq.connections import OpenAI as OpenAIConnection, E2B as E2BConnection
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.tools.e2b_sandbox import E2BInterpreterTool

# 初始化 E2B 工具
e2b_tool = E2BInterpreterTool(
    connection=E2BConnection(api_key="E2B_API_KEY")
)

# 设置你的 LLM
llm = OpenAI(
    id="openai",
    connection=OpenAIConnection(api_key="OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.3,
    max_tokens=1000,
)

# 创建 Agent
agent = Agent(
    name="react-agent",
    llm=llm, # 语言模型实例
    tools=[e2b_tool],  # Agent 可使用的工具列表
    role="Senior Data Scientist",  # Agent 的角色
    max_loops=10, # 处理循环的最大次数
)

async def run_async_agent():
    # 使用输入异步运行 Agent
    result = await agent.run(
        input_data={
            "input": "Add the first 10 numbers and tell if the result is prime.",
        }
    )

    print(result.output.get("content"))


# 执行异步函数
if __name__ == "__main__":
    asyncio.run(run_async_agent())

使用 WorkFlow 配置两个并行 Agent

from dynamiq import Workflow
from dynamiq.nodes.llms import OpenAI
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes.agents import Agent

# 设置你的 LLM
llm = OpenAI(
    connection=OpenAIConnection(api_key="OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.1,
)

# 定义第一个 Agent：问答 Agent
first_agent = Agent(
    name="Expert Agent",
    llm=llm,
    role="Professional writer with the goal of producing well-written and informative responses.",
    id="agent_1",
    max_loops=5
)

# 定义第二个 Agent：诗歌创作者
second_agent = Agent(
    name="Poetic Rewriter Agent",
    llm=llm,
    role="Professional writer with the goal of rewriting user input as a poem without changing its meaning.",
    id="agent_2",
    max_loops=5
)


# 创建一个工作流以使用相同的输入运行这两个 Agent
# `Workflow` 类简化了管道中一系列节点的设置和执行。
# 它会在可能的情况下自动并行运行 Agent。
wf = Workflow()
wf.flow.add_nodes(first_agent)
wf.flow.add_nodes(second_agent)

# 定义工作流的等效替代方式：
# from dynamiq.flows import Flow
# wf = Workflow(flow=Flow(nodes=[agent_first, agent_second]))

# 使用输入运行工作流
result = wf.run(
    input_data={"input": "How are sin(x) and cos(x) connected in electrodynamics?"},
)

# 打印两个 Agent 的输入和输出
print('--- Agent 1: Input ---\n', result.output[first_agent.id].get("input").get('input'))
print('--- Agent 1: Output ---\n', result.output[f"agent_1"].get("content"))