File size: 7,919 Bytes
a23394a | 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 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | #!/usr/bin/env python
from __future__ import annotations
import argparse
import asyncio
import os
import re
import io
import contextlib
from typing import Optional
from dataflow_agent.state import DFRequest, DFState
from dataflow_agent.workflow.wf_pipeline_write import create_operator_write_graph
from dataflow_agent.utils import get_project_root
PROJDIR = get_project_root()
def parse_args():
p = argparse.ArgumentParser(description="Run operator flow: match -> write -> (optional debug loop)")
p.add_argument('--chat-api-url', default='http://123.129.219.111:3000/v1/', help='LLM Chat API base')
p.add_argument('--model', default='gpt-4o', help='LLM model name')
p.add_argument('--language', default='en', help='Prompt output language')
p.add_argument('--target', required=True, help='User requirement / purpose for new operator')
p.add_argument('--category', default='Default', help='Operator category for matching (fallback if no classifier)')
p.add_argument('--output', default='', help='Optional path to write generated operator code')
p.add_argument('--json-file', default='', help='Path to test jsonl file used in debug run')
p.add_argument('--need-debug', action='store_true', help='Enable debug loop for executing and fixing the operator')
p.add_argument('--max-debug-rounds', type=int, default=3, help='Max debug rounds when --need-debug is set')
return p.parse_args()
"""
Entry script for operator write workflow.
Only keeps CLI parsing, DFState creation, and running the workflow graph.
Graph construction and node/tool definitions live in workflow/wf_pipeline_write.py
"""
async def main():
args = parse_args()
from dataflow.cli_funcs.paths import DataFlowPath
req = DFRequest(
language=args.language,
chat_api_url=args.chat_api_url,
api_key=os.getenv("DF_API_KEY", "sk-dummy"),
model=args.model,
target=args.target,
need_debug=bool(args.need_debug),
max_debug_rounds=int(args.max_debug_rounds),
# 默认使用 dataflowagent 下的 10 条测试数据
json_file=(args.json_file or f"{PROJDIR}/tests/test.jsonl"),
)
state = DFState(request=req, messages=[])
if args.output:
state.temp_data["pipeline_file_path"] = args.output
# 若用户通过参数提供了类别，也存到 temp_data 作为兜底
if args.category:
state.temp_data["category"] = args.category
# 显式初始化调试轮次
state.temp_data["round"] = 0
graph = create_operator_write_graph().build()
# LangGraph 默认 recursion_limit=25，当 need_debug 且 max_debug_rounds 较大时容易超限。
# 计算一个保守上限：主链 4 步 + 每轮 5 步 * 轮次 + buffer 5。
recursion_limit = 4 + 5 * int(args.max_debug_rounds) + 5
final_state: DFState = await graph.ainvoke(state, config={"recursion_limit": recursion_limit})
# ---- 打印结果摘要 ----
print("==== Match Operator Result ====")
try:
if isinstance(final_state, dict):
matched = final_state.get("matched_ops")
if not matched:
matched = (
final_state.get("agent_results", {})
.get("match_operator", {})
.get("results", {})
.get("match_operators", [])
)
else:
matched = getattr(final_state, "matched_ops", [])
if not matched and hasattr(final_state, "agent_results"):
matched = (
final_state.agent_results.get("match_operator", {})
.get("results", {})
.get("match_operators", [])
)
print("Matched ops:", matched or [])
except Exception:
print("Matched ops: <unavailable>")
print("\n==== Writer Result ====")
try:
# 兼容 dict 与 DFState 两种返回形态
if isinstance(final_state, dict):
code_str = (
final_state.get("temp_data", {}).get("pipeline_code", "")
or final_state.get("draft_operator_code", "")
or final_state.get("agent_results", {}).get("write_the_operator", {}).get("results", {}).get("code", "")
)
if not code_str:
fp = final_state.get("temp_data", {}).get("pipeline_file_path")
if fp:
from pathlib import Path
p = Path(fp)
try:
if p.exists():
code_str = p.read_text(encoding="utf-8")
except Exception:
pass
else:
code_str = (
getattr(final_state, "temp_data", {}).get("pipeline_code", "")
or getattr(final_state, "draft_operator_code", "")
or getattr(getattr(final_state, "agent_results", {}), "get", lambda *_: {})("write_the_operator", {}).get("results", {}).get("code", "")
)
if not code_str:
fp = getattr(final_state, "temp_data", {}).get("pipeline_file_path")
if fp:
from pathlib import Path
p = Path(fp)
try:
if p.exists():
code_str = p.read_text(encoding="utf-8")
except Exception:
pass
except Exception:
code_str = ""
print(f"Code length: {len(code_str)}")
if args.output:
print(f"Saved to: {args.output}")
else:
# 为避免终端刷屏，仅展示前 1000 字符
preview = (code_str or "")[:1000]
print("Code preview:\n", preview)
# ---- Debug runtime 移至 workflow 的 instantiate_operator_main_node ----
# 入口脚本不再负责内联实例化执行，避免过度复杂。
# ---- 执行结果摘要 ----
# 汇总执行结果（鲁棒回退）：execution_result -> agent_results -> 挂载属性
# 兼容 dict 与 DFState 两种返回形态
if isinstance(final_state, dict):
exec_res = final_state.get("execution_result", {}) or {}
if not exec_res or ("success" not in exec_res):
exec_res = final_state.get("agent_results", {}).get("operator_executor", {}).get("results", {}) or exec_res
else:
exec_res = getattr(final_state, "execution_result", {}) or {}
if (not exec_res or ("success" not in exec_res)) and hasattr(final_state, "agent_results"):
exec_res = final_state.agent_results.get("operator_executor", {}).get("results", {}) or exec_res
success = bool(exec_res.get("success"))
print("\n==== Execution Result (instantiate) ====")
print("Success:", success)
if not success:
stderr = (exec_res.get("stderr") or exec_res.get("traceback") or "")
print("stderr preview:\n", (stderr or "")[:500])
# ---- 调试实例化输出预览（来自 instantiate_operator_main_node） ----
try:
dbg = None
if isinstance(final_state, dict):
dbg = (final_state.get("temp_data") or {}).get("debug_runtime")
else:
dbg = getattr(final_state, "temp_data", {}).get("debug_runtime")
if dbg:
print("\n==== Debug Runtime Preview ==== ")
ik = dbg.get("input_key")
ak = dbg.get("available_keys")
print("input_key:", ik)
if ak:
print("available_keys:", ak)
stdout_pv = (dbg.get("stdout") or "")[:1000]
stderr_pv = (dbg.get("stderr") or "")[:1000]
if stdout_pv:
print("[debug stdout]\n", stdout_pv)
if stderr_pv:
print("[debug stderr]\n", stderr_pv)
except Exception:
pass
if __name__ == "__main__":
asyncio.run(main())
|