| """ |
| 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.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: |
| """ |
| Workflow factory: dfa run --wf test_graph |
| """ |
| builder = GenericGraphBuilder(state_model=TestGraphState, |
| entry_point="test_graph") |
|
|
| |
| |
| |
| |
| @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的火车票!!!!!!!!!!!!!" |
|
|
|
|
| |
|
|
| |
| |
| |
| 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: 主状态对象 |
| """ |
| |
| state.agent_results["step2"] = {"msg": "hello step2"} |
| |
| |
| |
| |
| |
| |
| return state |
|
|
| |
| |
| |
| nodes = { |
| "test_graph": test_graph_node, |
| "step2": step2, |
| '_end_': lambda state: state, |
| } |
|
|
| |
| |
| |
| edges = [ |
| ("test_graph", "step2"), |
| ("step2", "_end_"), |
| ] |
|
|
| builder.add_nodes(nodes).add_edges(edges) |
| return builder |
