File size: 5,216 Bytes
d82bbe4 | 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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """
test_graph workflow
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
生成时间: 2025-12-01 20:16:43
1. 在 **TOOLS** 区域定义需要暴露给 Prompt 的前置工具
2. 在 **NODES** 区域实现异步节点函数 (await-able)
3. 在 **EDGES** 区域声明有向边
4. 最后返回 builder.compile() 或 GenericGraphBuilder
"""
from __future__ import annotations
import json
from dataclasses import Field
from pydantic import BaseModel
from dataflow_agent.states.test_graph_state import TestGraphState
# from dataflow_agent.state import TestGraphState
from dataflow_agent.graphbuilder.graph_builder import GenericGraphBuilder
from dataflow_agent.workflow.registry import register
from dataflow_agent.agentroles import (
create_agent,
create_simple_agent,
create_react_agent,
create_graph_agent,
create_vlm_agent,
SimpleConfig,
ReactConfig,
GraphConfig,
VLMConfig,
ExecutionMode,
)
from dataflow_agent.toolkits.tool_manager import get_tool_manager
from langchain.tools import tool
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from dataflow_agent.graphbuilder.graph_builder import GenericGraphBuilder
from dataflow_agent.logger import get_logger
log = get_logger(__name__)
@register("test_graph")
def create_test_graph_graph() -> GenericGraphBuilder: # noqa: N802
"""
Workflow factory: dfa run --wf test_graph
"""
builder = GenericGraphBuilder(state_model=TestGraphState,
entry_point="test_graph") # 自行修改入口
# ----------------------------------------------------------------------
# TOOLS (pre_tool definitions)
# ----------------------------------------------------------------------
# 例:
@builder.pre_tool("purpose", "test_graph")
def _purpose(state: TestGraphState):
return "请问日期 11-29 的天气是什么？？？如果天气晴朗，请帮我购买这天的火车票！！"
@builder.post_tool("test_graph")
@tool
def _get_tomorrow_weather(data_str: str):
"""
获取明天天气
Args:
data_str: 日期字符串，格式为 "MM-DD"
"""
return "明天天气晴朗!!!!!!!!!!!!!"
@builder.post_tool("test_graph")
@tool
def _get_ticket(data_str: str):
"""
购买日期的火车票
Args:
data_str: 日期字符串，格式为 "MM-DD"
"""
return "购买1张11-29的火车票!!!!!!!!!!!!!"
# ----------------------------------------------------------------------
# ==============================================================
# NODES
# ==============================================================
async def test_graph_node(state: TestGraphState) -> TestGraphState:
"""
示例节点 1: 使用新的策略模式创建和执行 Agent
新版 Agent 创建方式推荐使用 `create_agent` 配合配置对象 (Config)
或使用便捷函数 `create_simple_agent`, `create_react_agent` 等。
执行模式说明:
- SimpleConfig: 简单模式，单次 LLM 调用
- ReactConfig: ReAct 模式，带验证和重试的循环
- GraphConfig: 图模式，用于执行带工具的子图 (LangGraph)
- VLMConfig: 视觉语言模型模式
"""
agent = create_graph_agent(
name="test_graph",
model_name="gpt-4o",
temperature=0.1,
max_tokens=16384,
parser_type="json",
)
state = await agent.execute(state=state)
log.critical(f"state.messages: {state.messages}")
# 可选：处理执行结果
agent_result = state.agent_results.get(agent.role_name, {})
log.info(f"Agent {agent.role_name} 执行结果: {agent_result}")
return state
async def step2(state: TestGraphState) -> TestGraphState:
"""
示例节点 2: 处理agent执行结果
Args:
state: 主状态对象
"""
# TODO: 替换为真正的业务逻辑
state.agent_results["step2"] = {"msg": "hello step2"}
# 示例：从 step1 的结果中提取数据
# if "code_reviewer" in state.agent_results:
# review_result = state.agent_results["code_reviewer"]
# # 处理审查结果...
return state
# ==============================================================
# 注册 nodes / edges
# ==============================================================
nodes = {
"test_graph": test_graph_node,
"step2": step2,
'_end_': lambda state: state, # 终止节点
}
# ------------------------------------------------------------------
# EDGES (从节点 A 指向节点 B)
# ------------------------------------------------------------------
edges = [
("test_graph", "step2"),
("step2", "_end_"), # 指向终止节点
]
builder.add_nodes(nodes).add_edges(edges)
return builder |