Dynamiq 是一个用于智能体（agentic）AI 和大语言模型（LLM）应用的编排框架

欢迎使用 Dynamiq！🤖

Dynamiq 是您的一站式生成式 AI（Gen AI）框架，旨在简化 AI 驱动应用的开发。Dynamiq 专注于编排检索增强生成（RAG, Retrieval-Augmented Generation）和大语言模型（LLM, Large Language Model）智能体（agents）。

快速开始

准备好了吗？以下是使用 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_template = """
将以下文本翻译成英文：{{ text }}
"""

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

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

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

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

带异步执行的简单 ReAct 智能体

一个可访问 E2B 代码解释器并能解决复杂编码任务的智能体。

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(
    name="react-agent",
    llm=llm, # 语言模型实例
    tools=[e2b_tool],  # 智能体可使用的工具列表
    role="高级数据科学家",  # 智能体的角色
    max_loops=10, # 处理循环次数上限
)

async def run_async_agent():
    # 异步运行智能体并传入输入
    result = await agent.run(
        input_data={
            "input": "将前 10 个数字相加，并判断结果是否为质数。",
        }
    )

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


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

使用 WorkFlow 配置两个并行智能体

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,
)

# 定义第一个智能体：问答智能体
first_agent = Agent(
    name="专家智能体",
    llm=llm,
    role="专业写手，目标是生成条理清晰且信息丰富的回答。",
    id="agent_1",
    max_loops=5
)

# 定义第二个智能体：诗歌重写智能体
second_agent = Agent(
    name="诗歌重写智能体",
    llm=llm,
    role="专业写手，目标是将用户输入重写为诗歌，同时不改变其原意。",
    id="agent_2",
    max_loops=5
)


# 创建一个工作流，以相同输入同时运行两个智能体
# `Workflow` 类简化了在流水线中设置和执行一系列节点的过程。
# 它会自动尽可能地并行运行智能体。
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": "在电动力学中，sin(x) 和 cos(x) 是如何关联的？"},
)

# 打印两个智能体的输入和输出
print('--- 智能体 1: 输入 ---\n', result.output[first_agent.id].get("input").get('input'))
print('--- 智能体 1: 输出 ---\n', result.output[first_agent.id].get("output").get('content'))
print('--- 智能体 2: 输入 ---\n', result.output[second_agent.id].get("input").get('input'))
print('--- 智能体 2: 输出 ---\n', result.output[second_agent.id].get("output").get('content'))

使用 WorkFlow 配置两个串行智能体

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

from dynamiq.nodes.node import InputTransformer, NodeDependency

设置你的 LLM（大语言模型）

llm = OpenAI(
    connection=OpenAIConnection(api_key="OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.1,
)

first_agent = Agent(
    name="Expert Agent",
    llm=llm,
    role="Professional writer with the goal of producing well-written and informative responses.",  # Agent 的角色
    id="agent_1",
    max_loops=5
)

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.",  # Agent 的角色
    id="agent_2",
    depends=[NodeDependency(first_agent)],  # 设置对第一个 agent 的依赖
    input_transformer=InputTransformer(
        selector={"input": f"${[first_agent.id]}.output.content"}  # 将第一个 agent 的输出提取为输入
    ),
    max_loops=5
)

# 创建一个工作流（Workflow），根据依赖关系依次运行 agents。
# 如果没有工作流，你需要手动先运行 `first_agent`，收集其输出，
# 然后再手动将该输出作为输入传递给 `second_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?"},
)

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

多智能体编排（Multi-agent orchestration）

from dynamiq import Workflow
from dynamiq.connections import OpenAI as OpenAIConnection, ScaleSerp as ScaleSerpConnection
from dynamiq.flows import Flow
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI
from dynamiq.nodes.tools.scale_serp import ScaleSerpTool
from dynamiq.nodes.types import Behavior, InferenceMode

llm = OpenAI(
    connection=OpenAIConnection(api_key="OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.1,
)

search_tool = ScaleSerpTool(connection=ScaleSerpConnection(api_key="SCALESERP_API_KEY"))

research_agent = Agent(
    name="Research Analyst",
    role="Find recent market news and provide referenced highlights.",
    llm=llm,
    tools=[search_tool],
    inference_mode=InferenceMode.XML,
    max_loops=6,
    behaviour_on_max_loops=Behavior.RETURN,
)

writer_agent = Agent(
    name="Brief Writer",
    role="Turn research highlights into a concise executive brief.",
    llm=llm,
    inference_mode=InferenceMode.XML,
    max_loops=4,
    behaviour_on_max_loops=Behavior.RETURN,
)

manager_agent = Agent(
    name="Manager",
    role=(
        "Delegate research and writing to sub-agents.\n"
        "Always call tools with {'input': '<task>'} payloads and assemble the final brief."
    ),
    llm=llm,
    tools=[research_agent, writer_agent],
    inference_mode=InferenceMode.XML,
    parallel_tool_calls_enabled=True,
    max_loops=8,
    behaviour_on_max_loops=Behavior.RETURN,
)

workflow = Workflow(flow=Flow(nodes=[manager_agent]))

result = workflow.run(
    input_data={"input": "Summarize the latest developments in battery technology for investors."},
)

print(result.output[manager_agent.id]["output"]["content"])

RAG（检索增强生成）- 文档索引流程

此工作流接收输入的 PDF 文件，对其进行预处理，转换为向量嵌入（vector embeddings），并存储到 Pinecone 向量数据库中。
此处示例使用的是 Pinecone 中已存在的索引。你可以在 docs/tutorials/rag 页面找到创建索引的示例。

from io import BytesIO

from dynamiq import Workflow
from dynamiq.connections import OpenAI as OpenAIConnection, Pinecone as PineconeConnection
from dynamiq.nodes.converters import PyPDFConverter
from dynamiq.nodes.splitters.document import DocumentSplitter
from dynamiq.nodes.embedders import OpenAIDocumentEmbedder
from dynamiq.nodes.writers import PineconeDocumentWriter

rag_wf = Workflow()

# PyPDF 文档转换器
converter = PyPDFConverter(document_creation_mode="one-doc-per-page")
rag_wf.flow.add_nodes(converter)  # 将节点添加到 DAG（有向无环图）

# 文档切分器
document_splitter = (
    DocumentSplitter(
        split_by="sentence",
        split_length=10,
        split_overlap=1,
    )
    .inputs(documents=converter.outputs.documents)  # 将 converter 节点的输出映射为当前节点的预期输入
    .depends_on(converter)
)
rag_wf.flow.add_nodes(document_splitter)

# OpenAI 向量嵌入
embedder = (
    OpenAIDocumentEmbedder(
        connection=OpenAIConnection(api_key="OPENAI_API_KEY"),
        model="text-embedding-3-small",
    )
    .inputs(documents=document_splitter.outputs.documents)
    .depends_on(document_splitter)
)
rag_wf.flow.add_nodes(embedder)

# Pinecone 向量存储
vector_store = (
    PineconeDocumentWriter(
        connection=PineconeConnection(api_key="PINECONE_API_KEY"),
        index_name="default",
        dimension=1536,
    )
    .inputs(documents=embedder.outputs.documents)
    .depends_on(embedder)
)
rag_wf.flow.add_nodes(vector_store)

# 准备输入的 PDF 文件
file_paths = ["example.pdf"]
input_data = {
    "files": [
        BytesIO(open(path, "rb").read()) for path in file_paths
    ],
    "metadata": [
        {"filename": path} for path in file_paths
    ],
}

# 运行 RAG 索引流程
rag_wf.run(input_data=input_data)

RAG - 文档检索流程

一个简单的检索型 RAG 流程，用于搜索相关文档，并利用检索到的文档回答用户的原始问题。

from dynamiq import Workflow
from dynamiq.connections import OpenAI as OpenAIConnection, Pinecone as PineconeConnection
from dynamiq.nodes.embedders import OpenAITextEmbedder
from dynamiq.nodes.retrievers import PineconeDocumentRetriever
from dynamiq.nodes.llms import OpenAI
from dynamiq.prompts import Message, Prompt

# 初始化 RAG 检索工作流
retrieval_wf = Workflow()

# 共享的 OpenAI 连接
openai_connection = OpenAIConnection(api_key="OPENAI_API_KEY")

# 用于查询嵌入的 OpenAI 文本嵌入器
embedder = OpenAITextEmbedder(
    connection=openai_connection,
    model="text-embedding-3-small",
)
retrieval_wf.flow.add_nodes(embedder)

# Pinecone 文档检索器
document_retriever = (
    PineconeDocumentRetriever(
        connection=PineconeConnection(api_key="PINECONE_API_KEY"),
        index_name="default",
        dimension=1536,
        top_k=5,
    )
    .inputs(embedding=embedder.outputs.embedding)
    .depends_on(embedder)
)
retrieval_wf.flow.add_nodes(document_retriever)

定义提示模板（prompt template）

prompt_template = """ 请根据提供的上下文回答问题。

问题: {{ query }}

上下文: {% for document in documents %}

• {{ document.content }} {% endfor %}

"""

使用 OpenAI LLM 生成答案

prompt = Prompt(messages=[Message(content=prompt_template, role="user")])

answer_generator = ( OpenAI( connection=openai_connection, model="gpt-4o", prompt=prompt, ) .inputs( documents=document_retriever.outputs.documents, query=embedder.outputs.query, ) # 从向量存储节点获取文档，从嵌入器（embedder）获取查询 .depends_on([document_retriever, embedder]) ) retrieval_wf.flow.add_nodes(answer_generator)

运行 RAG（检索增强生成）检索流程

question = "发票中提供了哪些明细项目？" result = retrieval_wf.run(input_data={"query": question})

answer = result.output.get(answer_generator.id).get("output", {}).get("content") print(answer)

### 带记忆功能的简单聊天机器人
一个使用 `Memory` 模块存储和检索对话历史的简单聊天机器人。

```python
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.memory import Memory
from dynamiq.memory.backends.in_memory import InMemory
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI

AGENT_ROLE = "乐于助人的助手，目标是提供有用的信息并回答问题"
llm = OpenAI(
    connection=OpenAIConnection(api_key="OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.1,
)

memory = Memory(backend=InMemory())

agent = Agent(
    name="Agent",
    llm=llm,
    role=AGENT_ROLE,
    id="agent",
    memory=memory,
)


def main():
    print("欢迎使用 AI 聊天！（输入 'exit' 退出）")
    while True:
        user_input = input("你: ")
        user_id = "user"
        session_id = "session"
        if user_input.lower() == "exit":
            break

        response = agent.run({"input": user_input, "user_id": user_id, "session_id": session_id})
        response_content = response.output.get("content")
        print(f"AI: {response_content}")


if __name__ == "__main__":
    main()

图编排器（Graph Orchestrator）

图编排器允许创建针对特定用例定制的任意架构。
以下是一个管理邮件反馈与迭代优化流程的简单工作流示例。

from typing import Any

from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes.agents.orchestrators.graph import END, START, GraphOrchestrator
from dynamiq.nodes.agents.orchestrators.graph_manager import GraphAgentManager
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI

llm = OpenAI(
    connection=OpenAIConnection(api_key="OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.1,
)

email_writer = Agent(
    name="email-writer-agent",
    llm=llm,
    role="根据反馈撰写个性化邮件。",
)


def gather_feedback(context: dict[str, Any], **kwargs):
    """收集关于邮件草稿的反馈。"""
    feedback = input(
        f"邮件草稿:\n"
        f"{context.get('history', [{}])[-1].get('content', '无草稿')}\n"
        f"输入 SEND 发送邮件，CANCEL 退出，或提供反馈以优化邮件：\n"
    )

    reiterate = True

    result = f"收集到的反馈: {feedback}"

    feedback = feedback.strip().lower()
    if feedback == "send":
        print("####### 邮件已发送！ #######")
        result = "邮件已发送！"
        reiterate = False
    elif feedback == "cancel":
        print("####### 邮件已取消！ #######")
        result = "邮件已取消！"
        reiterate = False

    return {"result": result, "reiterate": reiterate}


def router(context: dict[str, Any], **kwargs):
    """根据提供的反馈决定下一个状态。"""
    if context.get("reiterate", False):
        return "generate_sketch"

    return END


orchestrator = GraphOrchestrator(
    name="Graph orchestrator",
    manager=GraphAgentManager(llm=llm),
)

# 将任务附加到状态。当相应状态被触发时，这些任务将被执行。
orchestrator.add_state_by_tasks("generate_sketch", [email_writer])
orchestrator.add_state_by_tasks("gather_feedback", [gather_feedback])

# 通过添加边（edges）定义状态之间的流程。
# 此配置创建了从 START -> "generate_sketch" -> "gather_feedback" 的状态序列。
orchestrator.add_edge(START, "generate_sketch")
orchestrator.add_edge("generate_sketch", "gather_feedback")

# 为 "gather_feedback" 状态添加条件边，使流程可根据条件进行分支。
# router 函数将决定流程应转向 "generate_sketch"（继续迭代）还是 END（结束流程）。
orchestrator.add_conditional_edge("gather_feedback", ["generate_sketch", END], router)


if __name__ == "__main__":
    print("欢迎使用邮件撰写工具。")
    email_details = input("请提供邮件详情：")
    orchestrator.run(input_data={"input": f"撰写并发送邮件，并提供关于邮件状态的反馈：{email_details}"})

贡献指南

我们非常欢迎各种贡献！无论是提交 bug 报告、功能请求，还是发起 pull request，请查看我们的 CONTRIBUTING.md 了解如何参与。

许可证

Dynamiq 是开源项目，采用 Apache 2.0 许可证 发布。

祝你编码愉快！🚀

Dynamiq 快速上手指南

Dynamiq 是一个用于编排智能体（Agentic AI）和大语言模型（LLM）应用的开源框架，特别适用于构建 RAG（检索增强生成）和多智能体系统。

---

环境准备

• 操作系统：支持 Linux、macOS 和 Windows
• Python 版本：3.10 或更高版本
• 依赖服务（按需）：
  • OpenAI API Key（用于调用 GPT 模型）
  • 其他工具 API（如 E2B、ScaleSerp、Pinecone 等，根据示例需求）

💡 建议使用虚拟环境（如 venv 或 conda）隔离依赖。

---

安装步骤

方式一：通过 PyPI 安装（推荐）

pip install dynamiq

若国内网络访问较慢，可尝试使用清华源加速：

pip install dynamiq -i https://pypi.tuna.tsinghua.edu.cn/simple

方式二：从源码安装（适用于开发或最新特性）

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

需提前安装 Poetry。

---

基本使用

以下是最简单的 LLM 调用示例，实现文本翻译功能：

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

# 定义提示模板
prompt_template = """
Translate the following text into English: {{ text }}
"""

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,
    prompt=prompt
)

# 执行推理
result = llm.run(
    input_data={
        "text": "Hola Mundo!"
    }
)

print(result.output)

✅ 将 OPENAI_API_KEY 替换为你自己的 OpenAI API 密钥即可运行。

更多高级用法（如多智能体、工作流、RAG 等）请参考 官方文档。